LangChain Tool Calling Demo

浏览器（带 Cookie）
  → FastAPI 路由（提取 request.headers["cookie"]）
    → LangChain Engine（注入 RunnableConfig.configurable）
      → @tool 函数（从 config 中读取 cookie，附加到 HTTP 请求头）
        → 外部服务（鉴权通过 ✅）

from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool

@tool
async def list_user_databases(
    page: int = 1,
    page_size: int = 20,
    *,
    config: RunnableConfig,        # ← LangChain 自动注入，不会暴露给 LLM
) -> str:
    """获取当前用户创建的数据库列表。"""

    # 从 config 中提取用户鉴权信息
    user_cookie = config.get("configurable", {}).get("user_cookie", "")

    # 将鉴权信息透传到外部服务请求中
    headers = {"Cookie": user_cookie}
    async with httpx.AsyncClient() as client:
        resp = await client.get("http://external-service/api/databases", headers=headers)
        return resp.text

from langchain_core.runnables import RunnableConfig

# 构建包含用户信息的 config
config: RunnableConfig = {
    "configurable": {
        "user_cookie": user_cookie,   # 从 HTTP 请求头中提取
        "user_token": user_token,     # 也可以传递 Bearer Token
    }
}

# 调用 LLM 时传入 config（LLM 本身不使用，但会透传给工具）
response = await llm_with_tools.ainvoke(messages, config=config)

# 执行工具时传入同一个 config → 工具函数就能拿到 user_cookie
tool_result = await tool_func.ainvoke(tool_args, config=config)

@router.post("/stream")
async def stream_message(req: SendMessageRequest, request: Request, ...):
    # 提取用户浏览器发来的原始 Cookie 字符串
    user_cookie = request.headers.get("cookie", "")

    # 传递给聊天引擎，最终会到达工具函数
    async for event in chat_stream(
        db=db, session_id=req.session_id,
        user_message=req.message,
        user_cookie=user_cookie,       # ← 透传
    ):
        yield f"data: {json.dumps(event)}\n\n"

用户: "帮我查一下我创建了哪些数据库"

  [第 1 轮] LLM 返回 tool_calls: [{name: "list_user_databases", args: {}}]
            → 执行工具，得到 JSON 结果
            → 将结果作为 ToolMessage 加入消息列表

  [第 2 轮] LLM 收到工具结果，生成最终文本回复
            → "您目前创建了 5 个数据库，分别是..."

@tool
async def my_new_tool(
    param1: str,
    param2: int = 10,
    *,
    config: RunnableConfig,
) -> str:
    """工具描述（LLM 会根据这段描述决定何时调用此工具）。"""
    user_cookie = config.get("configurable", {}).get("user_cookie", "")

    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://your-service.com/api/xxx",
            headers={"Cookie": user_cookie},
            params={"param1": param1, "param2": param2},
        )
        return resp.text

ALL_TOOLS = [
    list_user_databases,
    get_database_detail,
    call_authenticated_api,
    my_new_tool,                # ← 新增
]

app/
├── main.py            # FastAPI 入口 + 生命周期管理
├── config.py          # 环境变量配置
├── database.py        # 异步 SQLAlchemy 数据库连接
├── models.py          # 数据模型（User / ChatSession / ChatMessage）
├── auth.py            # JWT 认证 + bcrypt 密码哈希
├── engine.py          # ★ LangChain 聊天引擎（工具调用循环 + 流式输出）
├── tools.py           # ★ 工具定义（鉴权透传的核心实现）
├── routes_auth.py     # 认证 API（注册 / 登录 / 登出）
└── routes_chat.py     # 聊天 API（会话管理 / 消息发送 / SSE 流式）

templates/
├── login.html         # 登录 / 注册页面
└── index.html         # 聊天页面

mock_server.py         # Mock 外部服务（用于测试工具调用 + 鉴权透传）

# 1. 安装依赖
pip install -r requirements.txt

# 2. 配置环境变量
cp .env.example .env
# 编辑 .env，填入 OPENAI_API_KEY 等配置

# 3. 启动 Mock 外部服务（可选，用于测试工具调用）
python3 mock_server.py

# 4. 启动主服务
python3 run.py

# 5. 访问 http://localhost:8000