LangChain agent server trên VPS: build internal AI tool tự host

User (Slack/Web)  -> REST API FastAPI    -> LangGraph orchestrator      -> LLM (Claude/GPT-4o/Local)        -> Tool 1: search_docs (RAG ChromaDB)        -> Tool 2: query_database (Postgres MCP)        -> Tool 3: fetch_jira_issue        -> Tool 4: lookup_employee_info      -> Memory (Redis conversation history)      -> Output: structured JSON response  <- Stream tokens via SSE

# VPS Cloud 40 (2GB RAM, 399k)dnf install python3.12 python3-pip -y mkdir -p /opt/ai-agent && cd /opt/ai-agentpython3 -m venv venvsource venv/bin/activate pip install fastapi uvicorn[standard] langchain langchain-anthropic     langchain-openai langchain-community langgraph chromadb     sentence-transformers psycopg2-binary redis sse-starlette     python-dotenv slack-sdk

# /opt/ai-agent/main.pyfrom fastapi import FastAPIfrom pydantic import BaseModelfrom langchain_anthropic import ChatAnthropicfrom langgraph.prebuilt import create_react_agentfrom tools import tool_setimport os app = FastAPI(title="Internal AI Agent") llm = ChatAnthropic(    model="claude-sonnet-4-7-20260520",    api_key=os.environ['ANTHROPIC_API_KEY'],    temperature=0) agent = create_react_agent(llm, tools=tool_set) class Query(BaseModel):    user_id: str    message: str    thread_id: str = "default" @app.post("/chat")async def chat(q: Query):    result = await agent.ainvoke(        {"messages": [("user", q.message)]},        config={"configurable": {"thread_id": q.thread_id}}    )    return {"response": result['messages'][-1].content} @app.get("/health")def health():    return {"status": "ok"}

# /opt/ai-agent/tools.pyfrom langchain_core.tools import toolimport psycopg2, requests, os @tooldef search_company_docs(query: str) -> str:    """Search internal company documents and policies. Use for HR policy, IT setup, SOP."""    from chromadb import HttpClient    client = HttpClient(host="localhost", port=8000)    col = client.get_collection("company_docs")    results = col.query(query_texts=[query], n_results=5)    return "nn".join([d for d in results['documents'][0]]) @tooldef query_employee_info(employee_email: str) -> dict:    """Look up employee info by email: department, manager, start date, position."""    conn = psycopg2.connect(os.environ['HR_DB_URL'])    cur = conn.cursor()    cur.execute("SELECT name, dept, manager, start_date, position FROM employees WHERE email=%s", (employee_email,))    row = cur.fetchone()    if not row:        return {"error": "Employee not found"}    return {"name": row[0], "department": row[1], "manager": row[2], "start_date": str(row[3]), "position": row[4]} @tooldef fetch_jira_issue(issue_key: str) -> dict:    """Fetch Jira issue details. Use for project status, bug info."""    r = requests.get(        f"https://your-domain.atlassian.net/rest/api/3/issue/{issue_key}",        auth=(os.environ['JIRA_USER'], os.environ['JIRA_TOKEN'])    )    return r.json() @tooldef run_sql_readonly(sql: str) -> list:    """Run read-only SQL query on analytics database. Only SELECT allowed."""    if not sql.strip().upper().startswith('SELECT'):        return {"error": "Only SELECT queries allowed"}    conn = psycopg2.connect(os.environ['ANALYTICS_DB_URL'])    cur = conn.cursor()    cur.execute(sql)    cols = [d[0] for d in cur.description]    return [dict(zip(cols, row)) for row in cur.fetchmany(100)] tool_set = [search_company_docs, query_employee_info, fetch_jira_issue, run_sql_readonly]

# /opt/ai-agent/index_docs.pyfrom langchain_community.document_loaders import (    DirectoryLoader, PyPDFLoader, UnstructuredMarkdownLoader)from langchain.text_splitter import RecursiveCharacterTextSplitterfrom langchain_community.embeddings import SentenceTransformerEmbeddingsfrom langchain_chroma import Chroma def index():    loader = DirectoryLoader('/data/company-docs', glob="**/*.md",                             loader_cls=UnstructuredMarkdownLoader)    docs = loader.load()    splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100)    chunks = splitter.split_documents(docs)     embed = SentenceTransformerEmbeddings(model_name="BAAI/bge-m3")    db = Chroma(        collection_name="company_docs",        embedding_function=embed,        persist_directory="/opt/ai-agent/chroma-data"    )    db.add_documents(chunks)    print(f"Indexed {len(chunks)} chunks") if __name__ == "__main__":    index()

from langgraph.checkpoint.redis import RedisSaver memory = RedisSaver.from_conn_string(    redis_url="redis://localhost:6379/0") agent = create_react_agent(    llm,    tools=tool_set,    checkpointer=memory  # auto save state per thread_id)

SYSTEM_PROMPT = """Bạn là trợ lý AI nội bộ của công ty TND, tên là Tina. Năng lực:- Trả lời câu hỏi về policy HR, IT, quy trình công ty.- Lookup thông tin nhân viên bằng email.- Fetch Jira issue cho update project.- Query SQL analytics database (chỉ read). Nguyên tắc:- Luôn trả lời tiếng Việt rõ ràng, ngắn gọn dưới 5 câu trừ khi cần liệt kê.- Nếu không biết, nói "Mình không tìm thấy thông tin, vui lòng hỏi anh/chị quản lý."- Không bịa data. Luôn cite source khi trả lời từ document.- Bảo mật: không lộ password, key, token trong response.- Tone: thân thiện, professional, không quá formal.""" agent = create_react_agent(    llm,    tools=tool_set,    checkpointer=memory,    state_modifier=SYSTEM_PROMPT)

# /etc/systemd/system/ai-agent.service[Unit]Description=AI Agent FastAPIAfter=network.target [Service]User=aiagentWorkingDirectory=/opt/ai-agentEnvironmentFile=/opt/ai-agent/.envExecStart=/opt/ai-agent/venv/bin/uvicorn main:app --host 127.0.0.1 --port 8001 --workers 2Restart=on-failure [Install]WantedBy=multi-user.target

# Caddyfileai.your-domain.com {    reverse_proxy 127.0.0.1:8001 {        flush_interval -1        transport http {            response_header_timeout 120s        }    }    basicauth /chat {        team $2a$14$your_caddy_hash    }}

@app.post("/slack/command")async def slack_command(req: Request):    form = await req.form()    user_id = form['user_id']    text = form['text']    channel = form['channel_id']     # Slack expects response in 3s, so async ack first    asyncio.create_task(process_and_respond(user_id, text, form['response_url']))    return {"response_type": "ephemeral", "text": "Tina đang nghĩ..."} async def process_and_respond(user_id, message, response_url):    result = await agent.ainvoke(        {"messages": [("user", message)]},        config={"configurable": {"thread_id": f"slack-{user_id}"}}    )    answer = result['messages'][-1].content    requests.post(response_url, json={"response_type": "in_channel", "text": answer})

from sse_starlette.sse import EventSourceResponse @app.post("/chat/stream")async def chat_stream(q: Query):    async def event_generator():        async for chunk in agent.astream(            {"messages": [("user", q.message)]},            config={"configurable": {"thread_id": q.thread_id}},            stream_mode="updates"        ):            for k, v in chunk.items():                if k == 'agent':                    content = v['messages'][-1].content                    if content:                        yield {"event": "token", "data": content}        yield {"event": "done", "data": ""}    return EventSourceResponse(event_generator())

# /opt/langfuse/docker-compose.ymlservices:  langfuse-server:    image: langfuse/langfuse:3    environment:      DATABASE_URL: postgresql://lf:lfpass@db:5432/langfuse      NEXTAUTH_SECRET: $(openssl rand -base64 32)      NEXTAUTH_URL: https://lf.your-domain.com    ports: ["3001:3000"]  db:    image: postgres:16    environment: { POSTGRES_USER: lf, POSTGRES_PASSWORD: lfpass, POSTGRES_DB: langfuse }    volumes: [lf-data:/var/lib/postgresql/data]volumes:  lf-data:

from langfuse.callback import CallbackHandlerlf = CallbackHandler(public_key="pk_lf_...", secret_key="sk_lf_...", host="https://lf.your-domain.com")result = await agent.ainvoke(input, config={"callbacks": [lf]})

from slowapi import Limiterfrom slowapi.util import get_remote_address limiter = Limiter(key_func=get_remote_address)app.state.limiter = limiter @app.post("/chat")@limiter.limit("20/minute")async def chat(request: Request, q: Query):    # check user budget remaining    daily_cost = redis_client.get(f"cost:{q.user_id}:{today}")    if daily_cost and float(daily_cost) > 2.0:        return {"error": "Daily budget exceeded (2 USD)"}    ...

# /opt/ai-agent/eval.pyimport jsonfrom agent import agent test_cases = [    {"q": "Chế độ nghỉ phép năm là gì?", "expected_tool": "search_company_docs", "expected_keyword": "12 ngày"},    {"q": "Email của manager phòng IT là gì?", "expected_tool": "query_employee_info"},    {"q": "Top 5 sản phẩm bán chạy tháng này?", "expected_tool": "run_sql_readonly"},] passed = 0for tc in test_cases:    result = agent.invoke({"messages": [("user", tc['q'])]})    if tc['expected_keyword'] in result['messages'][-1].content:        passed += 1        print(f"PASS: {tc['q']}")    else:        print(f"FAIL: {tc['q']}") print(f"n{passed}/{len(test_cases)} passed")

from langchain.cache import RedisSemanticCachefrom langchain.globals import set_llm_cache set_llm_cache(RedisSemanticCache(    redis_url="redis://localhost:6379/1",    embedding=embed,    score_threshold=0.95))