简介:本资源是O’Reilly Media出版的早期预发布电子书《Building Applications with AI Agents》,作者Michael Albada,面向AI工程实践者、系统架构师及对多智能体系统开发感兴趣的中高级开发者。资源聚焦AI Agent的设计原理与工程落地,重点覆盖Agent技能构建、任务编排(Orchestration)、记忆机制等核心模块,帮助读者在正式版发行前掌握前沿多代理系统开发方法论,适用于自动化服务、智能调度、安全编排等真实场景。压缩包为单文件PDF格式,大小558KB,内容含已发布的第3章(Skills)与第4章(Orchestration)等早期章节,结构清晰、代码与架构图虽未最终定稿但具备直接参考价值。目前已有441人学习下载,所含Early Release版本虽存在编辑未完善之处,但附有明确勘误提示与开源合规说明,便于读者结合实践同步验证、迭代理解。
1. 这不是“调用API”的说明书,而是一份能让你亲手把AI代理从“会聊天”变成“能干活”的技能工程实操笔记
你有没有试过让一个大模型查天气、算账、翻文档、改配置——结果它嘴上说得天花乱坠,手却一动不动?不是模型不行,是它缺“手”。这本书《Building Applications with AI Agents》第3章“Skills”,讲的正是这双“手”的设计逻辑与落地方法:技能(Skill)不是功能封装,而是AI代理在真实世界中执行动作的最小可验证单元。它把“让模型思考”和“让模型做事”彻底拆开——思考交给LLM,动作交给Skill,再用工具调用(tool calling)机制桥接二者。这种分层,直接决定了你的AI应用是停留在Demo阶段,还是能进生产环境跑通闭环。适合三类人:正在用LangChain/LlamaIndex搭Agent但总卡在“调不动外部服务”的开发者;想把已有业务系统(如CRM、ERP、监控平台)快速接入AI工作流的后端工程师;以及被“智能体幻觉”反复背刺、急需用确定性模块对冲不确定性风险的技术负责人。它不讲抽象架构图,只讲怎么写一个@tool装饰器、怎么绑定到llm.bind_tools()、怎么从tool_calls里安全提取参数、怎么处理API失败时的fallback路径——全是我在某跨平台系统中踩过坑、修过半夜、上线后稳跑三个月的血泪经验。
2. 技能的本质:从“函数封装”到“可编排的动作单元”
2.1 为什么不能直接在prompt里写“请调用天气API”?
这是新手最常掉的第一个坑。LLM本质是文本生成器,它没有操作系统权限,无法发起HTTP请求、读写数据库、执行shell命令。所谓“调用API”,其实是两段独立代码的协作:LLM负责决策(该不该调?调哪个?参数是什么?),而Skill负责执行(发请求、校验响应、抛异常、返回结构化数据)。如果跳过Skill层,直接让LLM“假装调用”,结果就是:输出里写着“已获取北京今日气温23℃”,但实际什么都没发生——这就是典型的“幻觉执行”。真正的技能必须满足三个硬性条件:有明确输入类型约束(pydantic)、有可捕获的错误路径(try/except)、有确定性返回值(非字符串)。下面这个计算器Skill看似简单,但每行都在为生产环境埋下可靠性伏笔:
from langchain_core.tools import tool from pydantic import BaseModel, Field class MultiplyInput(BaseModel): x: float = Field(description="第一个乘数") y: float = Field(description="第二个乘数") @tool("multiply", args_schema=MultiplyInput) def multiply(x: float, y: float) -> float: """将x与y相乘,返回浮点结果""" try: result = x * y # 防止科学计数法干扰后续解析 if abs(result) < 1e-6 or abs(result) > 1e10: return round(result, 6) return result except OverflowError: raise ValueError(f"乘法溢出:{x} * {y} 超出浮点数范围") except Exception as e: raise ValueError(f"乘法执行失败:{str(e)}")注意:这里用了
args_schema强制类型校验,而非仅靠函数签名。LLM生成的tool_call参数若为字符串"393"和"12.25",LangChain会自动尝试转换为float;若转换失败,则整个调用链终止,不会传入函数内部导致TypeError。这是防御性编程的第一道闸门。
2.2 手写技能 vs API技能:什么时候该自己写,什么时候该调别人?
书中把技能分为“Local Skills”和“API-Based Skills”,但实际选型远比二分法复杂。我总结了一张决策表,基于某图像处理Demo的真实迭代过程:
| 场景特征 | 推荐方案 | 理由 | 血泪教训 |
|---|---|---|---|
| 输入输出格式固定、计算逻辑确定(如单位换算、日期加减) | 手写Local Skill | 无网络依赖、毫秒级响应、100%可控 | 曾用Wikipedia API查“闰年规则”,因API限流导致用户等待超15秒,后全量替换为本地calendar.isleap() |
| 需实时数据且第三方API稳定(如股票、汇率) | 封装API-Based Skill | 避免重复造轮子,利用专业服务商的数据更新能力 | 初期自建汇率爬虫,因目标网站反爬升级,三天内失效两次,运维成本远超API费用 |
| 数据源在内网、无标准API(如公司OA审批流) | 手写Local Skill + 内网HTTP Client | 避免暴露内网服务到公网,符合安全审计要求 | 试图用LangChain的RequestsGetTool直连OA,被安全部门叫停,最终改用Flask封装成内网微服务再注册为Skill |
| 模型本身擅长但需结构化输出(如从文本抽手机号) | 用LLM+OutputParser替代Skill | 减少系统耦合,避免为简单NLP任务引入额外服务 | 为“提取邮箱”单独写Skill,结果发现GPT-4o的JSON mode比正则更准,且无需维护 |
关键结论:Skill不是越多越好,而是越少越稳。每个Skill都意味着一个故障点、一次网络IO、一条异常链路。能用LLM原生能力解决的,绝不外包给Skill;能用本地计算解决的,绝不依赖外部API。
2.3 技能注册的隐藏陷阱:bind_tools()之后,LLM真的“看见”了你的工具吗?
很多开发者卡在这一步:写了@tool,也bind_tools([tool])了,但LLM始终不触发调用。根本原因在于——LLM的工具感知能力极度依赖工具描述质量。书中示例的"""Multiply 'x' times 'y'."""过于简略,实际生产中必须遵循三要素原则:
- 动词开头:用“获取”“查询”“创建”“更新”等强动作词,禁用“用于”“可以”等弱表达
- 明确主体与宾语:
get_stock_price(ticker: str)→获取指定股票代码的最新交易价格 - 标注关键约束:
ticker需为大写英文,如"AAPL",不支持加密货币代码
实测对比:当描述从"Get stock price"升级为"获取指定美股代码的实时收盘价(仅支持NYSE/NASDAQ上市股票,ticker需为大写,如'AAPL'或'MSFT')",调用成功率从68%提升至94%。这不是玄学,是LLM对自然语言指令的token级理解偏差——它需要足够多的上下文锚点来定位工具边界。
3. 技能链路的完整闭环:从LLM决策到结果注入的七步实操
3.1 完整调用链:为什么invoke()要调两次?
书中代码只展示了单次llm_with_tools.invoke(messages),但真实场景中,一次LLM调用最多触发一个Skill,而复杂任务往往需要多跳(multi-hop)。比如“查北京天气→若温度<10℃→推荐穿羽绒服”,就必须:
- LLM第一次调用,决定查天气并生成
tool_call - 执行天气Skill,拿到JSON响应
- 将响应作为
ToolMessage追加到messages - LLM第二次调用,基于新消息决定是否推荐衣物
- 若需推荐,则触发第二个Skill(如穿搭建议API)
- 再次注入结果
- 最终LLM合成自然语言回复
这个七步链路不可省略,否则就会出现“查完天气就结束,不继续推理”的断链。下面是某客户支持系统的生产级实现:
from langchain_core.messages import HumanMessage, ToolMessage, AIMessage from langchain_openai import ChatOpenAI # 初始化带工具的LLM llm = ChatOpenAI(model_name="gpt-4o", temperature=0.3) llm_with_tools = llm.bind_tools([get_weather, recommend_clothes]) def run_agent(query: str) -> str: messages = [HumanMessage(content=query)] # 第一次调用:LLM决定是否需要工具 ai_msg = llm_with_tools.invoke(messages) messages.append(ai_msg) # 检查是否需要调用工具 if ai_msg.tool_calls: for tool_call in ai_msg.tool_calls: try: # 执行工具(此处简化,实际需根据tool_call['name']路由) if tool_call['name'] == "get_weather": tool_result = get_weather.invoke(tool_call) elif tool_call['name'] == "recommend_clothes": tool_result = recommend_clothes.invoke(tool_call) # 注入工具结果,注意:必须用ToolMessage,不能用AIMessage tool_msg = ToolMessage( content=tool_result.content, name=tool_call['name'], tool_call_id=tool_call['id'] # 关键!必须匹配LLM生成的id ) messages.append(tool_msg) except Exception as e: # 工具执行失败时,注入错误消息,让LLM自行处理 error_msg = f"工具{tool_call['name']}执行失败:{str(e)}" messages.append(ToolMessage( content=error_msg, name=tool_call['name'], tool_call_id=tool_call['id'] )) # 第二次调用:LLM基于工具结果生成最终回复 final_response = llm_with_tools.invoke(messages) return final_response.content # 测试 print(run_agent("北京现在穿什么衣服合适?"))逻辑说明:
tool_call_id必须严格匹配LLM生成的id,否则LangChain无法关联工具结果与原始调用。这是框架级约定,不是可选项。参数temperature=0.3而非0,是为了在确定性与推理灵活性间平衡——纯0会导致LLM拒绝多跳推理,死守“只调一次”。
3.2 工具结果注入的三种模式:何时用ToolMessage,何时用AIMessage?
很多开发者混淆这两者,导致LLM无法正确解析工具返回。核心区别:
ToolMessage:必须用于注入Skill执行结果,name字段对应工具名,tool_call_id必须与tool_call['id']一致,content应为结构化数据(如JSON字符串或dict)AIMessage:仅用于LLM自身输出,绝不能用于注入工具结果
错误示范(会导致LLM忽略工具结果):
# ❌ 错误:用AIMessage注入 messages.append(AIMessage(content=str(tool_result)))正确示范:
# ✅ 正确:用ToolMessage注入,且携带完整元数据 tool_msg = ToolMessage( content=json.dumps({"temp": 12.5, "condition": "cloudy"}), # 建议转JSON字符串 name="get_weather", tool_call_id=tool_call['id'] ) messages.append(tool_msg)参数说明:
content建议序列化为JSON字符串而非Python dict,因为LangChain内部对dict的序列化行为不稳定;tool_call_id是LangChain v0.1+的强制字段,缺失将导致工具结果丢失。
3.3 多工具并发调用:如何让LLM一次触发多个Skill?
书中示例均为单工具调用,但真实场景常需并发(如同时查股价+查新闻)。LangChain原生支持,但需满足两个前提:
- LLM模型必须支持并行工具调用(GPT-4o、Claude-3 Opus均支持,GPT-3.5不支持)
bind_tools()传入的工具列表必须包含所有可能被并发调用的工具
# 注册多个工具 tools = [get_stock_price, get_news_summary, get_company_info] # 绑定时一次性传入全部 llm_with_tools = llm.bind_tools(tools) # LLM可自主决定并发调用 query = "分析苹果公司(AAPL)当前股价、最新财报新闻及主营业务" messages = [HumanMessage(query)] ai_msg = llm_with_tools.invoke(messages) # ai_msg.tool_calls 可能包含3个元素,按顺序执行即可 for tool_call in ai_msg.tool_calls: # ... 执行对应工具避坑提示:并发不等于并行——LangChain默认串行执行
tool_calls。若需真正并行,需用asyncio.gather()包装,但必须确保所有工具函数为async。同步工具强行await会阻塞事件循环。
4. 避坑:五个让AI代理在生产环境凌晨三点给你打电话的真实问题
4.1 现象:LLM反复调用同一个工具,参数完全相同,形成无限循环
原因:工具返回内容未提供足够信息供LLM终止推理。例如天气Skill返回{"temp": 12.5},但LLM需要“是否低于10℃”这个布尔判断才能决定下一步,而它无法从数字12.5自行推导出“否”。
解决:在Skill内部完成业务逻辑判断,返回语义化结果。
@tool def get_weather_decision(city: str) -> str: """获取城市天气决策:返回'cold'/'warm'/'hot',直接用于下游分支""" temp = get_weather_raw(city) # 内部调用原始天气API if temp < 10: return "cold" elif temp > 25: return "hot" else: return "warm"4.2 现象:工具调用成功,但LLM返回“我无法回答”或空响应
原因:ToolMessage.content为None或空字符串,LangChain将其视为空消息,LLM缺乏推理依据。
解决:强制工具返回非空内容,即使失败也要返回错误描述。
try: result = requests.get(url).json() return json.dumps(result) # ✅ 非空JSON字符串 except Exception as e: return f"ERROR: {str(e)}" # ✅ 明确错误标识4.3 现象:本地Skill在开发机运行正常,部署到Docker后报ModuleNotFoundError
原因:@tool装饰器注册的函数在模块顶层,但Docker镜像中未安装对应依赖(如requests、pandas)。
解决:在Skill函数内部做依赖检查,失败时抛出可读异常。
@tool def process_csv(file_path: str) -> str: try: import pandas as pd except ImportError: raise ImportError("Skill 'process_csv' requires pandas. Please install with 'pip install pandas'") # ... 后续逻辑4.4 现象:API-Based Skill偶发超时,但LLM仍认为调用成功
原因:requests.get()默认无超时,网络抖动时线程挂起,LLM等待超时后放弃,但Skill未抛出异常。
解决:所有HTTP请求必须设timeout=(3, 7)(连接3秒,读取7秒),并捕获requests.exceptions.Timeout。
try: response = requests.get(url, timeout=(3, 7)) response.raise_for_status() return response.json() except requests.exceptions.Timeout: raise ValueError("API请求超时,请检查网络或重试") except requests.exceptions.HTTPError as e: raise ValueError(f"API返回HTTP错误:{e}")4.5 现象:多个Skill共享全局状态(如数据库连接池),高并发下连接耗尽
原因:手写Skill中使用模块级变量(如db_conn = create_connection()),在异步环境下被所有协程共享。
解决:Skill函数必须无状态,连接池等资源通过依赖注入或工厂函数获取。
# ❌ 错误:模块级连接 db_conn = psycopg2.connect(...) @tool def query_db(sql: str): return db_conn.execute(sql) # 并发时崩溃 # ✅ 正确:每次调用新建连接(轻量API)或用连接池(重负载) @tool def query_db(sql: str): with get_db_connection() as conn: # 工厂函数返回新连接 return conn.execute(sql)5. 生产就绪:技能可观测性与降级策略的实战配置
5.1 给每个Skill加上黄金指标监控
不能只看“调用成功”,要监控三个黄金指标:
- 成功率:
success_count / total_count,阈值<95%告警 - P95延迟:排除超时请求后的95分位响应时间,阈值>2s告警
- 错误分布:按
ValueError、ConnectionError、Timeout分类统计,定位根因
用Prometheus + Grafana实现(某高校实验平台部署方案):
from prometheus_client import Counter, Histogram import time # 定义指标 SKILL_CALL_COUNTER = Counter( 'skill_call_total', 'Total number of skill calls', ['skill_name', 'status'] # status: success/fail ) SKILL_LATENCY_HISTOGRAM = Histogram( 'skill_latency_seconds', 'Skill execution latency', ['skill_name'] ) @tool def get_weather(city: str) -> str: start_time = time.time() try: # ... 实际逻辑 result = _fetch_weather_from_api(city) SKILL_CALL_COUNTER.labels(skill_name="get_weather", status="success").inc() return result except Exception as e: SKILL_CALL_COUNTER.labels(skill_name="get_weather", status="fail").inc() raise e finally: latency = time.time() - start_time SKILL_LATENCY_HISTOGRAM.labels(skill_name="get_weather").observe(latency)提示:
status标签必须区分success/fail,不能只用exception_type,因为部分业务异常(如“城市不存在”)应计入成功率,而网络异常必须告警。
5.2 降级策略:当天气API崩了,如何不让整个Agent瘫痪?
书中未提降级,但生产环境必须有Plan B。我们采用三级降级:
- 一级降级(缓存):用Redis缓存最近1小时天气,TTL=3600,命中则直接返回
- 二级降级(静态数据):缓存失效时,返回预置的“北京:四季分明,当前季节平均温度”静态文案
- 三级降级(LLM兜底):前两级均失败,让LLM基于历史知识生成合理推测(如“北京冬季通常寒冷,建议穿厚外套”)
@tool def get_weather_fallback(city: str) -> str: # 一级:Redis缓存 cache_key = f"weather:{city}" cached = redis_client.get(cache_key) if cached: return json.loads(cached) # 二级:静态数据 static_data = get_static_weather(city) if static_data: redis_client.setex(cache_key, 3600, json.dumps(static_data)) return static_data # 三级:LLM兜底(调用另一个轻量LLM) fallback_llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0) prompt = f"基于地理常识,描述{city}当前季节的典型天气和着装建议,用中文,50字内" return fallback_llm.invoke([HumanMessage(content=prompt)]).content5.3 技能版本管理:如何安全灰度发布新Skill?
当你要替换旧版get_stock_price为支持港股的新版时,不能一刀切。我们用A/B测试路由:
# 注册两个版本 @tool("get_stock_price_v1") def get_stock_price_v1(ticker: str) -> float: # 旧逻辑 @tool("get_stock_price_v2") def get_stock_price_v2(ticker: str) -> float: # 新逻辑,支持港股 # 在LLM调用前,动态选择工具 def get_active_tools() -> List: # 从配置中心读取灰度比例 rollout_rate = config.get("stock_skill_rollout", 0.0) if random.random() < rollout_rate: return [get_stock_price_v2, get_weather] # 新版+其他工具 else: return [get_stock_price_v1, get_weather] # 旧版+其他工具 # 每次请求动态绑定 llm_with_tools = llm.bind_tools(get_active_tools())参数说明:
rollout_rate从Consul配置中心动态拉取,支持秒级生效。新旧版本共存期间,日志中打标version=v1或version=v2,便于对比成功率与延迟。
从那以后我每次上线新Skill,都强制走一遍这三步:先加Prometheus埋点,再配降级兜底,最后设灰度开关。不是为了炫技,是某次凌晨三点被报警电话叫醒后,我盯着监控面板上那个突兀的红色success_rate: 23%曲线,终于明白——AI代理的可靠性,不取决于它多聪明,而取决于你为它的每一次“动手”准备了多少条退路。希望帮到你。
本文还有配套的精品资源,点击获取