Docker环境下构建AI Agent：从基础实现到LangGraph框架-AI智能范式网

Docker环境下构建AI Agent：从基础实现到LangGraph框架

艾弥儿

1. 项目概述

在当今AI技术快速发展的背景下，构建能够自主思考和行动的AI Agent已成为开发者关注的热点。本文将详细介绍如何在Docker环境中从零开始构建一个基础AI Agent，重点讲解容器化交互与Prompt工程的核心技术。

这个项目基于WSL2 + Ubuntu 24.04 + Docker Desktop构建的云原生开发平台，所有服务（包括大模型、数据库等）均以独立容器形式运行并通过Docker Compose统一编排。我们将从最基础的原理入手，逐步实现一个能够理解用户需求、调用工具并给出合理响应的微型Agent系统。

2. 环境准备与基础架构

2.1 开发环境配置

我们的开发环境采用以下技术栈：

WSL2 (Windows Subsystem for Linux)
Ubuntu 24.04
Docker Desktop
Python 3.12.8

这种配置的优势在于：

保持了Windows系统的易用性
获得了Linux环境的开发便利
通过Docker实现了服务隔离和便捷部署

提示：具体环境配置可参考相关技术博客，本文重点将放在Agent开发的核心逻辑上。

2.2 Docker网络架构

在容器化开发中，网络配置是第一个需要解决的问题。我们的架构中，各个服务（Qwen大模型、Redis、MySQL）都运行在独立的容器中，通过Docker Compose进行编排。

关键概念：Docker Compose DNS

当运行docker-compose up时，Docker会自动创建一个内部网络
在这个网络中，服务名就是域名
例如，如果compose文件中大模型服务名为qwen-local，那么容器内部访问地址应为http://qwen-local:8000

常见错误：

python复制# ❌ 错误写法（容器内部无法访问宿主机localhost）
http://localhost:8000/v1/chat/completions

# ✅ 正确写法（使用服务名作为域名）
http://qwen-local:8000/v1/chat/completions

2.3 Python虚拟环境配置

为了隔离项目依赖，我们使用Python虚拟环境。配置步骤如下：

创建虚拟环境：

bash复制python -m venv .venv

激活虚拟环境：

bash复制# Windows
.venv\Scripts\activate

# Linux/macOS
source .venv/bin/activate

使用pyproject.toml管理依赖（替代传统的requirements.txt）：

toml复制[project]
name = "docker-agent-lab"
version = "0.1.0"
dependencies = [
    "requests>=2.31.0",
    "python-dotenv>=1.0.0",
    "ipython>=8.20.0",
]

安装依赖：

bash复制pip install --upgrade pip
pip install -e .

3. 基础Agent实现

3.1 Agent核心原理

Agent的本质是一个循环执行的过程，称为ReAct（Reasoning and Acting）循环：

观察(Observation): 接收用户输入
思考(Thought): 分析需要采取的行动
行动(Action): 调用适当的工具
结果(Result): 获取工具返回结果
循环: 直到问题解决

这个循环使Agent能够像人类一样逐步解决问题，而不是简单地一次性给出答案。

3.2 工具定义与注册

工具(Tools)是Agent与现实世界交互的接口。我们首先定义两个基础工具：

python复制from datetime import datetime

def get_current_date(*args, **kwargs) -> str:
    """获取当前的日期（包含年月日和星期几）"""
    now = datetime.now()
    date_str = now.strftime("%Y-%m-%d")
    weekday_str = now.strftime("%A")
    weekday_map = {
        "Monday": "星期一", "Tuesday": "星期二", "Wednesday": "星期三",
        "Thursday": "星期四", "Friday": "星期五", "Saturday": "星期六", 
        "Sunday": "星期日"
    }
    cn_weekday = weekday_map.get(weekday_str, weekday_str)
    return f"日期：{date_str}, 星期：{cn_weekday} ({weekday_str})"

def get_current_time(*args, **kwargs) -> str:
    """获取当前的具体时间（时:分:秒）"""
    now = datetime.now()
    return f"时间：{now.strftime('%H:%M:%S')}"

然后创建工具注册表，方便Agent查找和调用：

python复制TOOLS_REGISTRY = {
    "get_current_date": {
        "description": "当用户询问日期、今天几号、今天是星期几时使用。不需要参数。",
        "function": get_current_date
    },
    "get_current_time": {
        "description": "当用户询问具体几点、当前时刻时使用。不需要参数。",
        "function": get_current_time
    }
}

3.3 LLM交互层

与大型语言模型(LLM)的交互是Agent的核心。我们封装了一个标准的调用函数：

python复制import json
import re
import requests

MODEL_NAME = "qwen-3-4b"
LLM_API_URL = "http://localhost:7575/v1/chat/completions"

SYSTEM_PROMPT = """
你是一个智能助手。你可以使用以下工具来回答问题。
你必须严格按照 JSON 格式回复，不要包含任何 Markdown 标记（如 ```json）。
格式如下：
{
    "thought": "你现在的思考过程，分析用户需要什么",
    "action": "工具名称 (如果没有工具可用，填 null)",
    "action_input": "工具的参数 (JSON 对象，如果没有填 {})"
}
"""

def call_llm(messages: list, tools_desc: str) -> Dict[str, Any]:
    full_system_prompt = SYSTEM_PROMPT + tools_desc
    
    payload = {
        "model": MODEL_NAME,
        "messages": [
            {"role": "system", "content": full_system_prompt},
            *messages
        ],
        "temperature": 0.05,  # 低温度保证逻辑稳定
        "stream": False
    }
    
    try:
        response = requests.post(LLM_API_URL, json=payload, timeout=30)
        response.raise_for_status()
        content = response.json()['choices'][0]['message']['content']
        
        # 清理响应内容
        content = re.sub(r'<think>.*?</think>', '', content, flags=re.DOTALL)
        if content.startswith("```json"):
            content = content[7:]
        if content.endswith("```"):
            content = content[:-3]
            
        return json.loads(content.strip())
    except Exception as e:
        print(f"LLM调用失败: {e}")
        return {"thought": "系统错误", "action": None}

3.4 ReAct循环实现

将上述组件整合，实现完整的Agent循环：

python复制def run_agent(user_query: str):
    messages = [{"role": "user", "content": user_query}]
    print(f"👤 用户: {user_query}")
    
    max_steps = 5  # 防止死循环
    step = 0
    
    while step < max_steps:
        step += 1
        print(f"\n--- 🔄 第 {step} 轮思考 ---")
        
        # 1. 请求LLM
        result = call_llm(messages, json.dumps(TOOLS_DEFINITION, ensure_ascii=False))
        thought = result.get("thought", "")
        action = result.get("action")
        action_input = result.get("action_input", {})
        
        print(f"🧠 思考: {thought}")
        
        # 2. 判断是否结束
        if not action:
            print(f"🤖 Agent: {thought}")
            break
        
        # 3. 执行工具
        if action in TOOLS_REGISTRY:
            tool_func = TOOLS_REGISTRY[action]["function"]
            print(f"🛠️ 执行工具: {action}")
            try:
                observation = tool_func(**action_input)
                print(f"📝 观察结果: {observation}")
                
                # 将结果反馈给LLM
                messages.append({"role": "assistant", "content": json.dumps(result, ensure_ascii=False)})
                messages.append({"role": "user", "content": f"工具 {action} 执行完毕，结果是：{observation}。请根据结果回答用户。"})
            except Exception as e:
                error_msg = f"工具执行出错: {str(e)}"
                print(f"❌ {error_msg}")
                messages.append({"role": "user", "content": error_msg})
        else:
            print(f"⚠️ 未知工具: {action}")
            messages.append({"role": "user", "content": f"错误：找不到工具 {action}。请重试。"})
            
    if step >= max_steps:
        print("⚠️ 达到最大思考步数，停止。")

3.5 测试运行

启动Agent并测试简单查询：

python复制if __name__ == "__main__":
    query = "现在几点了？顺便告诉我今天是星期几。"
    run_agent(query)

预期输出示例：

code复制👤 用户: 现在几点了？顺便告诉我今天是星期几。

--- 🔄 第 1 轮思考 ---
🧠 思考: 用户需要当前时间以及今天的星期几信息，我需要调用两个工具来获取这些数据。
🛠️ 执行工具: get_current_time
📝 观察结果: 时间：22:49:13

--- 🔄 第 2 轮思考 ---
🧠 思考: 用户需要当前时间及星期几，已通过get_current_time获取时间，需调用get_current_date获取星期几后整合回答。
🛠️ 执行工具: get_current_date
📝 观察结果: 日期：2026-02-26, 星期：星期四 (Thursday)

--- 🔄 第 3 轮思考 ---
🧠 思考: 已获取当前时间22:49:13和日期信息（2026-02-26，星期四），现在可以整合信息回答用户。
🤖 Agent: 已获取当前时间22:49:13和日期信息（2026-02-26，星期四），现在可以整合信息回答用户。

4. 进阶：使用LangGraph框架重构

4.1 为什么需要框架？

虽然手写Agent有助于理解底层原理，但在实际项目中会面临诸多挑战：

复杂的状态管理
并发处理
错误恢复
可扩展性

LangGraph是LangChain团队推出的专门用于构建Agent的状态机框架，它提供了更结构化的方式来管理Agent的工作流。

4.2 环境升级

首先更新依赖：

toml复制dependencies = [
    "langchain>=0.3.0",
    "langchain-community>=0.3.0",
    "langgraph>=0.2.0",
    "langchain-openai>=0.2.0",
]

4.3 使用LangGraph重构Agent

4.3.1 定义工具

使用LangGraph的@tool装饰器定义工具：

python复制from langchain_core.tools import tool

@tool
def get_current_date() -> str:
    """获取当前的日期和星期几。适用于询问今天周几、日期的问题。"""
    now = datetime.now()
    return f"日期：{now.strftime('%Y-%m-%d')}, 星期：{now.strftime('%A')}"

@tool
def get_current_time() -> str:
    """获取当前的具体时间（时:分:秒）。适用于询问几点、剩余时间的问题。"""
    return f"时间：{datetime.now().strftime('%H:%M:%S')}"

tools = [get_current_date, get_current_time]

4.3.2 定义状态

python复制from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

4.3.3 构建节点

python复制from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode

# 初始化模型
llm = ChatOpenAI(
    model="qwen-3-4b",
    base_url="http://localhost:7575/v1",
    api_key="not-needed",
    temperature=0.1
)

# 绑定工具
llm_with_tools = llm.bind_tools(tools)

# 定义模型节点
def call_model(state: AgentState):
    messages = state["messages"]
    raw_response = llm_with_tools.invoke(messages)
    extracted_tool_calls = extract_tool_calls_from_content(raw_response.content)
    
    response = AIMessage(
        content=raw_response.content,
        tool_calls=extracted_tool_calls
    )
    
    return {"messages": [response]}

# 工具节点
tool_node = ToolNode(tools)

4.3.4 构建图与路由逻辑

python复制from langgraph.graph import StateGraph, START, END

def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if isinstance(last_message, AIMessage) and last_message.tool_calls:
        return "tools"
    return END

workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
    "agent", 
    should_continue,
    {
        "tools": "tools",
        END: END
    }
)
workflow.add_edge("tools", "agent")

app = workflow.compile()

4.3.5 运行Agent

python复制def run_framework_agent(query: str):
    print(f"👤 用户：{query}")
    inputs = {"messages": [HumanMessage(content=query)]}
    
    for event in app.stream(inputs, stream_mode="values"):
        last_msg = event["messages"][-1]
        role = last_msg.type
        content = last_msg.content
        
        if role == "ai":
            if last_msg.tool_calls:
                print(f"🧠 思考: 准备调用工具 {[t['name'] for t in last_msg.tool_calls]}")
            else:
                print(f"🤖 Agent: {content}")
        elif role == "tool":
            print(f"📝 工具结果: {content}")

if __name__ == "__main__":
    query = "现在几点了？顺便告诉我今天是星期几。"
    run_framework_agent(query)

5. 常见问题与解决方案

5.1 工具调用协议不兼容

某些开源模型可能不支持标准的OpenAI工具调用协议。解决方案是手动解析模型输出并注入工具调用信息：

python复制import re,json

def extract_tool_calls_from_content(content: str):
    tool_calls = []
    matches = re.findall(r"<tool_call>\s*({.*?})\s*</tool_call>", content, re.DOTALL)
    for i, match in enumerate(matches):
        try:
            data = json.loads(match.strip())
            name = data.get("name")
            arguments = data.get("arguments", {})
            if name:
                tool_calls.append({
                    "name": name,
                    "args": arguments,
                    "id": f"call_{i}_{name}",
                    "type": "tool_call"
                })
        except Exception as e:
            print(f"解析工具调用失败: {e}")
    return tool_calls

5.2 Docker网络连接问题

确保容器间通信使用正确的服务名而非localhost。可以通过以下命令检查服务状态：

bash复制docker-compose ps

5.3 模型响应不稳定

可以通过调整以下参数改善：

降低temperature值（如0.1）
优化prompt设计
增加max_steps限制防止死循环

6. 性能优化建议

批量处理工具调用：当Agent需要调用多个工具时，可以并行执行以提高效率
缓存常用工具结果：对于频繁使用的工具（如时间查询），可以添加短期缓存
异步IO：使用异步请求提高并发性能
状态持久化：将重要状态保存到数据库，支持断点续传

7. 扩展思路

这个基础Agent可以进一步扩展为：

多Agent协作系统：多个Agent分工合作解决复杂问题
长期记忆：集成向量数据库存储历史对话和知识
动态工具加载：根据需要动态加载和卸载工具
可视化监控：添加对Agent决策过程的实时监控

通过本文介绍的方法，开发者可以构建出功能强大且灵活的AI Agent系统。关键在于理解底层原理，然后根据实际需求选择合适的抽象层级和框架。