2025 年已经过去，Chatbot 经过几年的发展已经改变了我们的生活工作，它擅长即时响应，提供比搜索引擎更直接的答案。正如 Similarweb 数据展示，Google 之外的 Chatbot 访问流量已不可忽视。

图 |  流量份额

Agent 的发展趋势

Chatbot 的普及不仅验证了 AI 交互的价值，更成为技术升级的铺垫。随着大语言模型（LLM）能力的持续突破，更具深度的 AI 智能体（Agent）加速走向应用 —— 它们能够模仿人类的思维逻辑与工作流程，系统性解决研究分析、报告撰写、PPT 制作等复杂工作任务。

过去一年，AI Coding Agent 发展尤为迅猛，已成为最具商业价值的细分领域，诸如 Cursor、Claude Code、Replit，lovable，CodeBuddy，Trae 等近百种工具涌现，催生出规模达数十亿美元的市场。

图 |  头部 AI 公司 ARR

热潮之下，Agent 的开发本质是一项复杂的系统性工程：

无论是短时记忆与长时记忆的协同管理，还是审慎调用外部工具为大语言模型（LLM）补充外部知识、完善其内置知识库，都需要投入大量研发精力。而作为 Agent 的核心组件，LLM 的选型与部署同样需综合考量多重因素，如不同模型的能力特点、成本控制、故障应对、速率限制，以及如何保障服务的连续性与稳定性等等...

为了解决上述问题，Agent 开发者通常这么选：

✅ 接入多家 MaaS 供应商

✅ 使用聚合平台提供的 API 服务

第一个方案需要兼容多个平台的协议，管理大量的支付账号、账单等。而第二个方案则免去了以上问题，只需一个账号即可访问所需模型。

OneThingAI 平台就为大家提供了这样的 API 服务，满足多种 LLM 模型的调用。其在实现上保证了 API 接入层和实际的推理服务间的分离。API 接入层运行在 CPU 集群，依托多年积累的成熟的软硬件技术，可实现远高于推理服务的稳定性。

Agent 的概念与设计模式

如上一章所言，Agent 是复杂工程。我们先给他下个定义：AI Agent 是一种能够感知环境、自主决策并执行动作的软件实体。其核心能力包括规划、工具使用、记忆和反思。不同于单次调用的 LLM，Agent 通过组合这些能力，可处理需多步骤推理的动态任务，如客户支持、数据分析或自动化流程。

然而，直接开发 Agent 系统容易导致代码臃肿、难以维护。设计模式通过提供可重用的架构模板，解决了以下问题：

• 复杂性管理：如将任务分解为可管理的组件，通过链式调用的方式组织顺序操作。

• 灵活性提升：如让一个 Agent 根据输入动态选择执行路径。

• 安全性保障：如防止有害输出或越权行为，清楚的设定 Agent 的能力边界。

• 效率优化：平衡成本与性能，如利用平台 cache hit 后 token 更便宜的特性来组织上下文。

以下我们参考《Agentic Design Patterns》整理一些简单模式：

1、链式执行模式

链式执行是 Agent 的基础模式，它将复杂任务分解为线性序列的子任务，每个子任务的输出作为下一个的输入。这种模式适用于有依赖关系的流程，如数据处理的提取-转换-加载（ETL）。其优势在于结构清晰、易于调试，但缺乏灵活性。

2、路由模式

路由模式使 Agent 能根据输入内容动态选择执行路径。例如，用户查询可能是信息请求或操作指令，路由器通过分析意图，将其分配给专用子 Agent（如搜索 Agent 或数据库 Agent）。这种模式提高了系统的适应性，尤其适合多领域任务。

3、工具使用模式

工具使用是Agent与外部世界交互的关键。通过函数调用（Function Calling），Agent 可以执行搜索、数据库查询或API 操作等任务。模式流程包括：定义工具描述 → LLM 决策调用 → 执行函数 → 整合结果。这打破了 LLM 的知识局限，使其能处理实时数据。例如，一个旅行规划智能体可以调用航班查询 API、酒店预订工具和地图服务，整合信息为用户生成完整行程。该模式是智能体与现实世界交互的关键。

4、并行化模式

并行化模式通过并发执行多个独立子任务来优化整体工作流的效率。该模式的核心在于识别任务中互不依赖的组成部分（如调用多个外部 API、处理不同数据块、生成内容的不同片段），使其同时运行，待所有子任务完成后再整合结果。例如，一个研究型智能体可以同时搜索不同来源的信息，而非顺序进行，从而显著减少等待时间。主流框架如 LangChain 通过RunnableParallel 结构实现并行执行，而 Google ADK 则提供 ParallelAgent 原语来协调多个子智能体的并发工作。

5、反思模式

反思模式为智能体引入了自我评估与迭代优化的能力。智能体在生成初步结果后，会基于预设标准或外部反馈对自身输出进行审查，识别潜在错误、不一致或可改进之处，并进行调整。例如，代码生成智能体在编写程序后，可运行单元测试或静态分析工具，根据结果反思并修正代码逻辑。这一模式显著提升了输出的可靠性与质量。

6、规划模式

规划模式要求智能体在执行任务前先制定一个详细的步骤计划。智能体需要理解最终目标，推理出达成目标所需的行动序列，考虑可能的分支和依赖关系，然后按计划逐步执行。例如，一个完成学术报告的智能体会先规划为：资料搜集、大纲制定、分章节撰写、图表生成、校对排版等步骤。规划模式适用于目标复杂、步骤间存在逻辑依赖的场景。

7、内存管理模式

Agent 需记忆历史交互以支持连续对话。内存模式分为短期内存（存储当前会话状态）和长期内存（持久化知识）。短期内存通常用会话状态管理，长期内存依赖向量数据库等外部存储。

以上 7 种模式组合后又可以形成一种新的模式，解决更复杂的场景问题。而随着模型技术的迭代演进，这些模式也需随之动态调整。更复杂的技术可以参考《Agentic Design Patterns》等。

Agent 的应用实践

*以下实践仅用于演示 Agent 工作流，不构成投资建议。

✅ 基于 Langchain 框架的实践

接下来，我们基于 Langchain 实现一个简单的股票分析 Agent。代码如下：

"""股票查询服务 - LangChain Agent"""import json, osfrom datetime import datetime, timedeltafrom typing import Optionalimport httpxfrom dotenv import load_dotenvfrom pydantic import SecretStrfrom langchain_core.tools import toolfrom langchain_core.messages import HumanMessagefrom langchain.agents import create_agentfrom langchain_openai import ChatOpenAIload_dotenv()STOCK_API = "https://push2his.eastmoney.com/api/qt/stock/kline/get"STOCK_MAP = {"茅台": "600519", "贵州茅台": "600519", "五粮液": "000858", "中国平安": "601318",             "招商银行": "600036", "工商银行": "601398", "腾讯": "00700", "阿里巴巴": "09988"}def get_market(code: str) -> str:    return "1" if code[:3] in ["600", "601", "603", "605", "688"] else "0"def query_stock(code: str, begin: Optional[str] = None, end: Optional[str] = None) -> dict:    """查询股票数据。默认最近1周，最多2年"""    now = datetime.now()    today = now.strftime("%Y%m%d")    # 默认最近1周，最多2年(500交易日)    if not begin:        begin = (now - timedelta(days=7)).strftime("%Y%m%d")    # 限制最大2年    two_years_ago = (now - timedelta(days=730)).strftime("%Y%m%d")    if begin < two_years_ago:        begin = two_years_ago    end_d = end or today    params = {"secid": f"{get_market(code)}.{code}", "klt": "101", "fqt": "1",              "fields1": "f1,f2,f3,f4,f5,f6", "fields2": "f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61",              "beg": begin, "end": end_d, "lmt": 500}    try:        resp = httpx.get(STOCK_API, params=params, timeout=10).json()        if resp.get("rc") != 0 or not resp.get("data"): return {"error": "未找到", "code": code}        data, klines = resp["data"], resp["data"].get("klines", [])        ipo = klines[0].split(",")[0].replace("-","") if klines else None        # begin早于IPO则调整        if ipo and begin < ipo: return query_stock(code, ipo, end)        # 返回摘要        latest = klines[-1].split(",") if klines else []        first = klines[0].split(",") if klines else []        result = {"code": code, "name": data.get("name",""), "count": len(klines),                  "range": f"{first[0]}~{latest[0]}" if first and latest else "",                  "latest": {"date": latest[0], "close": latest[2], "change": latest[8]+"%"} if latest else {}}        if len(klines) > 1:            prices = [float(k.split(",")[2]) for k in klines]            result.update({"high": round(max(prices),2), "low": round(min(prices),2), "avg": round(sum(prices)/len(prices),2)})        return result    except Exception as e: return {"error": str(e), "code": code}def get_code(name: str) -> Optional[str]:    if name in STOCK_MAP: return STOCK_MAP[name]    for n, c in STOCK_MAP.items():        if n in name or name in n: return c    return None@tooldef search_stock(name: str) -> str:    """股票名转代码"""    return get_code(name) or "未找到"@tooldef get_stock_price(code: str, begin: Optional[str] = None, end: Optional[str] = None) -> str:    """查股价。begin/end格式YYYYMMDD，不传默认最近1周，最多2年"""    return json.dumps(query_stock(code, begin, end), ensure_ascii=False)@tooldef analyze_stock(data: str) -> str:    """分析股票数据"""    try:        d = json.loads(data) if isinstance(data, str) else data        if "error" in d: return f"错误: {d['error']}"        lt = d.get("latest", {})        s = f"{d.get('name','')}({d.get('code','')}): {lt.get('close','?')}元, 涨跌{lt.get('change','?')}"        s += f"
区间: {d.get('range','')}, 共{d.get('count',0)}天"        if d.get("count", 0) > 1:            s += f"
最高{d.get('high')}, 最低{d.get('low')}, 均价{d.get('avg')}"            cur, avg = float(lt.get("close",0)), d.get("avg",0)            s += " [高位]" if cur > avg*1.1 else " [低位]" if cur < avg*0.9 else " [正常]"        return s    except Exception as e: return str(e)# === Agent (LangChain 1.0) ===api_key = os.getenv("API_KEY", "")llm = ChatOpenAI(model="deepseek-v3.2", base_url="https://api-model.onethingai.com/v2/openai",                 api_key=SecretStr(api_key) if api_key else None, temperature=0.6)tools = [search_stock, get_stock_price, analyze_stock]agent = create_agent(model=llm, tools=tools, system_prompt="""股票助手。步骤:1. search_stock找代码2. get_stock_price查价格(可选begin/end,格式YYYYMMDD,不传默认最近1周)3. analyze_stock分析用户若指定时间如"最近1个月"则计算对应begin""")# === FastAPI ===if __name__ == "__main__":    from fastapi import FastAPI    from pydantic import BaseModel    import uvicorn    app = FastAPI(title="股票AI", version="2.0")    class Q(BaseModel): query: str    @app.post("/chat")    def chat(q: Q):        result = agent.invoke({"messages": [HumanMessage(content=q.query)]})        messages = result.get("messages", [])        return {"answer": messages[-1].content if messages else ""}    @app.get("/")    def root(): return {"usage": "POST /chat: {'query':'分析茅台的股价'}"}    print("http://localhost:3000/docs")    uvicorn.run(app, host="0.0.0.0", port=3000)

要在 Langchain 中使用 OneThingAI 平台提供的 API，可以使用 Langchain_openai 这个库的 ChatOpenAI 类指定 base_url 为：https://api-model.onethingai.com/v2/openai

llm = ChatOpenAI(model="deepseek-v3.2", base_url="https://api-model.onethingai.com/v2/openai",                 api_key=SecretStr(api_key) if api_key else None, temperature=0.6)

注意：以上代码需要在 python 3.10.0 以上环境中运行，可以使用 pyenv 来管理虚拟环境，参考以下命令：

pyenv install 3.10.0pyenv local 3.10.0python -m venv venvsource venv/bin/activatepip install  langchain langchain_core langchain_openai dotenv pydantic uvicorn fastapi   

以上代码启动后即可通过访问本地 3000 端口，实现股票的分析：

curl -X POST -H "Content-Type: application/json" http://localhost:3000/chat  -d '{"query": "分析茅台的股价"}'

Agent 返回的 Markdown 渲染效果如下：

✅ 基于 Google ADK 框架的实践

基于 Google ADK 实现一个简单多 Agent 协作模式。root agent 代码如下：

"""根代理 - 协调股票查询和分析子代理 (Google ADK)"""from google.adk.agents import Agentfrom google.adk.models.lite_llm import LiteLlmfrom stock_agent.agents.stock_query_agent import stock_query_agentfrom stock_agent.agents.analysis_agent import analysis_agentfrom stock_agent.config.settings import ROOT_MODEL, API_KEY, BASE_URL# 使用 LiteLLM 支持 OpenAI 兼容 API (gpt-5.1)root_llm = LiteLlm(    model=ROOT_MODEL,    api_key=API_KEY,    api_base=BASE_URL,)# Root Agent 使用子代理作为工具root_agent = Agent(    name="stock_root_agent",    model=root_llm,    instruction="""你是股票查询和分析的总协调者。执行流程:1. 先用stock_query_agent查询股票代码和价格数据2. 将查询结果交给analysis_agent进行分析3. 整合结果返回给用户确保查询成功后再进行分析，如果查询失败直接返回错误信息。""",    sub_agents=[stock_query_agent, analysis_agent],)async def invoke(query: str) -> str:    """执行查询: Root Agent协调两个子代理"""    from google.adk.runners import Runner    from google.adk.sessions import InMemorySessionService    from google.genai import types    session_service = InMemorySessionService()    session = await session_service.create_session(        app_name="stock_agent",        user_id="user",        session_id="session"    )    runner = Runner(        agent=root_agent,        app_name="stock_agent",        session_service=session_service    )    content = types.Content(        role="user",        parts=[types.Part(text=query)]    )    final_response = ""    async for event in runner.run_async(        user_id="user",        session_id="session",        new_message=content    ):        if event.is_final_response() and event.content and event.content.parts:            final_response = event.content.parts[0].text    return final_response or "查询失败"

查询股票信息的 Agent 如下：

"""股票查询子代理 - 负责查找股票代码和获取价格数据 (Google ADK)"""from google.adk.agents import Agentfrom google.adk.models.lite_llm import LiteLlmfrom stock_agent.config.settings import STOCK_QUERY_MODEL, API_KEY, BASE_URLfrom stock_agent.tools import search_stock_tool, get_stock_price_tool# 使用 LiteLLM 支持 OpenAI 兼容 API (deepseek-v3.2)stock_query_llm = LiteLlm(    model=STOCK_QUERY_MODEL,    api_key=API_KEY,    api_base=BASE_URL,)stock_query_agent = Agent(    name="stock_query_agent",    model=stock_query_llm,    instruction="""你是股票查询专家。任务:1. 用search_stock把股票名称转为代码2. 用get_stock_price查询价格数据(可选begin/end,格式YYYYMMDD)只返回JSON数据,不做分析""",    tools=[search_stock_tool, get_stock_price_tool],)

分析股票信息的 Agent 如下：

"""分析子代理 - 负责分析股票数据 (Google ADK)"""from google.adk.agents import Agentfrom google.adk.models.lite_llm import LiteLlmfrom stock_agent.config.settings import ANALYSIS_MODEL, API_KEY, BASE_URLfrom stock_agent.tools import analyze_stock_tool# 使用 LiteLLM 支持 OpenAI 兼容 API (kimi-k2)analysis_llm = LiteLlm(    model=ANALYSIS_MODEL,    api_key=API_KEY,    api_base=BASE_URL,)analysis_agent = Agent(    name="analysis_agent",    model=analysis_llm,    instruction="""你是股票分析专家。任务:用analyze_stock分析股票数据JSON,给出:- 当前价格和涨跌- 区间统计(最高/最低/均价)- 判断当前处于高位/低位/正常返回简洁的分析结论""",    tools=[analyze_stock_tool],)

Google ADK 要能使用 OneThingAI 提供的 API，需通过 LiteLlm 实现。对比 Langchain ，这里的模型名需要 openai / 前缀来告诉 LiteLLM 使用 OpenAI 格式。

# OpenAI 兼容 API 配置API_KEY = os.getenv("API_KEY", "")BASE_URL = os.getenv("BASE_URL", "https://api-model.onethingai.com/v2/openai")# 模型配置 - 使用 LiteLLM 格式支持 OpenAI 兼容 API# 需要 openai/ 前缀告诉 LiteLLM 使用 OpenAI 格式# Root Agent 使用 deepseek-v3ROOT_MODEL = os.getenv("ROOT_MODEL", "openai/deepseek-v3")# Stock Query Agent 使用 deepseek-v3.2STOCK_QUERY_MODEL = os.getenv("STOCK_QUERY_MODEL", "openai/deepseek-v3.2")# Analysis Agent 使用 deepseek-v3.2ANALYSIS_MODEL = os.getenv("ANALYSIS_MODEL", "openai/deepseek-v3.2")

以上只举例了两个框架，Agent 框架还包含 claude code sdk，crew.ai，autogen 等等，如有需要后续补充。ADK 项目的完整代码可关注：https://github.com/OneThingAI/StockAgent