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, os
from datetime import datetime, timedelta
from typing import Optional
import httpx
from dotenv import load_dotenv
from pydantic import SecretStr
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
load_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
@tool
def search_stock(name: str) -> str:
"""股票名转代码"""
return get_code(name) or "未找到"
@tool
def 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)
@tool
def 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"\n区间: {d.get('range','')}, 共{d.get('count',0)}天"
if d.get("count", 0) > 1:
s += f"\n最高{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.0
pyenv local 3.10.0
python -m venv venv
source venv/bin/activate
pip 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 Agent
from google.adk.models.lite_llm import LiteLlm
from stock_agent.agents.stock_query_agent import stock_query_agent
from stock_agent.agents.analysis_agent import analysis_agent
from 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 Agent
from google.adk.models.lite_llm import LiteLlm
from stock_agent.config.settings import STOCK_QUERY_MODEL, API_KEY, BASE_URL
from 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 Agent
from google.adk.models.lite_llm import LiteLlm
from stock_agent.config.settings import ANALYSIS_MODEL, API_KEY, BASE_URL
from 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-v3
ROOT_MODEL = os.getenv("ROOT_MODEL", "openai/deepseek-v3")
# Stock Query Agent 使用 deepseek-v3.2
STOCK_QUERY_MODEL = os.getenv("STOCK_QUERY_MODEL", "openai/deepseek-v3.2")
# Analysis Agent 使用 deepseek-v3.2
ANALYSIS_MODEL = os.getenv("ANALYSIS_MODEL", "openai/deepseek-v3.2")

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

学AI大模型的正确顺序,千万不要搞错了

🤔2026年AI风口已来!各行各业的AI渗透肉眼可见,超多公司要么转型做AI相关产品,要么高薪挖AI技术人才,机遇直接摆在眼前!

有往AI方向发展,或者本身有后端编程基础的朋友,直接冲AI大模型应用开发转岗超合适!

就算暂时不打算转岗,了解大模型、RAG、Prompt、Agent这些热门概念,能上手做简单项目,也绝对是求职加分王🔋

在这里插入图片描述

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料,手把手帮你快速入门!👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型(GPT、文心一言等)特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架(LangChain等)实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块,看似清晰好上手,实则每个部分都有扎实的核心内容需要吃透!

我把大模型的学习全流程已经整理📚好了!抓住AI时代风口,轻松解锁职业新可能,希望大家都能把握机遇,实现薪资/职业跃迁~

这份完整版的大模型 AI 学习资料已经上传CSDN,朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费

在这里插入图片描述

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐