AI Agent开发实战：从零构建可交互Demo的完整指南

1次阅读

没有评论

共计 2206 个字符，预计需要花费 6 分钟才能阅读完成。

作为刚接触 AI Agent 开发的新手，我最初面临的最大困扰是：明明理解了大模型的基础用法，但当需要让 AI 主动调用工具、处理多轮对话时，代码立刻变得支离破碎。典型问题包括：

意图与执行割裂：大模型能识别用户想查天气，但不知道如何调用天气 API
状态管理混乱：每次对话都当成独立请求，无法实现类似『上一条说的城市』的上下文引用
工具调用不可控：多个工具并行时出现资源竞争，或某个工具超时导致整个流程阻塞

调研主流框架后，我的对比结论如下：

LangChain
优点：Python 生态完善，工具链丰富（内置 70+ 工具），调试可视化好
缺点：异步支持较弱，大型 Agent 性能下降明显
Semantic Kernel
优点：微软系技术栈友好，C#/Python 双支持，适合企业级应用
缺点：学习曲线陡峭，社区案例较少
纯 OpenAI API
优点：绝对控制权，适合研究底层机制
缺点：所有轮子都要自己造

新手推荐组合：LangChain + Pydantic（类型校验）+ FastAPI（轻量服务化）

from typing import List, Dict, Any
from pydantic import BaseModel
from langchain.agents import initialize_agent
from langchain.llms import OpenAI

class AgentCore:
    def __init__(self, tools: List[Any]):
        self.llm = OpenAI(temperature=0)
        self.agent = initialize_agent(
            tools, 
            self.llm,
            agent="react",  # 使用 ReAct 决策模式
            verbose=True
        )

    async def run(self, input_text: str, session_id: str) -> Dict:
        try:
            return await self.agent.arun(input=input_text)
        except Exception as e:
            return {"error": str(e)}

演示一个天气查询工具的完整实现：

from langchain.tools import tool
import requests

@tool
def get_weather(city: str) -> str:
    """查询指定城市天气，参数格式如：北京"""
    try:
        # 示例 API，实际需替换为真实服务
        resp = requests.get(f"https://weatherapi.example.com?city={city}", timeout=3)
        return resp.json()["forecast"]
    except TimeoutError:
        return "天气服务超时，请稍后重试"
    except Exception:
        return "无法获取天气信息"

# 注册工具
agent = AgentCore(tools=[get_weather])

使用 Redis 维护对话历史（简化版）：

import redis
from datetime import timedelta

r = redis.Redis()

def store_context(session_id: str, query: str, response: str):
    r.setex(f"agent:{session_id}:last_query", 
        timedelta(minutes=10),
        query
    )
    r.rpush(f"agent:{session_id}:history", response)

当多个工具需要共享资源（如数据库连接）时：

使用 asyncio.Lock 控制并发
为每个工具设置独立连接池

from asyncio import Lock
db_lock = Lock()

async def safe_db_query(sql: str):
    async with db_lock:
        return await db.execute(sql)

全局设置超时拦截器
实现熔断模式

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(2))
async def call_external_api(url: str):
    async with httpx.AsyncClient(timeout=3) as client:
        return await client.get(url)

当 Demo 需要转向生产环境时：