Function Calling（现在 OpenAI 叫 Tool Use）是让 LLM 从"聊天机器人"变成"能干活的 Agent"的关键能力。 核心原理 Function Calling 的本质是：你告诉模型"你可以调用这些函数"，模型在需要时会输出一个结构化的"我要调用 X 函数，参数是 Y"，然后由你的代码真正去执行这个函数，把结果再传给模型。 模型本身不执行任何代码，它只负责"决策"。 最简示例 from openai import OpenAI import json client = OpenAI() # 定义工具 tools = [ { "type": "function", "function": { "name": "get_weather", "description": "获取指定城市的天气", "parameters": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称，如：北京、上海" } }, "required": ["city"] } } } ] # 实际的工具函数 def get_weather(city: str) -> str: # 这里接真实天气 API return f"{city}今天晴，25°C" messages = [{"role": "user", "content": "北京今天天气怎么样？"}] response = client.chat.completions.create( model="gpt-4o", messages=messages, tools=tools, tool_choice="auto" ) # 判断是否要调用工具 message = response.choices[0].message if message.tool_calls: tool_call = message.tool_calls[0] args = json.loads(tool_call.function.arguments) # 执行工具 result = get_weather(**args) # 把结果传回给模型 messages.append(message) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result }) final_response = client.chat.completions.create( model="gpt-4o", messages=messages ) print(final_response.choices[0].message.content) 多工具调度 模型可以在一次回复里调用多个工具（parallel tool calling）： ...

Prompt 工程这个词听起来很高大上，但它本质上就是"怎么跟模型说话才能让它给出你想要的结果"。这篇文章分享一些实际项目里积累的经验。 明确角色和背景 给模型一个清晰的角色定位： # 模糊 "帮我写一段代码" # 清晰 "你是一个 Python 高级工程师，专注于性能优化。 帮我优化以下函数，要求： 1. 保持功能不变 2. 减少不必要的循环 3. 添加类型注解 4. 代码风格遵循 PEP 8" 角色描述告诉模型应该以什么视角和知识储备来回答。 Few-shot：给例子比给说明更有效 告诉模型你想要什么格式，最好的方式是给例子： 将以下文本分类为：正面/负面/中性 示例： - "这个产品质量很好" → 正面 - "一般般，没什么特别" → 中性 - "完全是浪费钱" → 负面 现在分类以下文本： - "还行，比预期好一点点" Few-shot 比描述规则更直观，模型更容易理解你的意图。 Chain of Thought：让模型"想一想再说" 对于需要推理的任务，让模型展示推理过程： # 直接问（效果差） "一个工厂每天生产 500 个零件，工人效率提升 20% 后， 需要多少天生产 18000 个零件？" # 加上 "一步一步思考"（效果好） "一个工厂每天生产 500 个零件，工人效率提升 20% 后， 需要多少天生产 18000 个零件？请一步一步计算。" “Let’s think step by step” 或"一步一步思考"这类指令能显著提升推理准确率，这是有论文支撑的结论。 ...

LangChain 的迭代速度极快，API 经常变，文档跟不上代码。这篇文章记录我踩过的一些有代表性的坑。 坑 1：版本兼容性 LangChain 把包拆分了，以前的 langchain 现在分成了： langchain-core：核心抽象 langchain：主包 langchain-community：第三方集成 langchain-openai、langchain-anthropic 等：各家模型的独立包 很多旧教程的 import 路径在新版本里已经不对了： # 旧写法（可能报 ImportError） from langchain.chat_models import ChatOpenAI # 新写法 from langchain_openai import ChatOpenAI 解决办法：固定版本，或者直接看报错信息里的迁移提示。 坑 2：ConversationBufferMemory 在 LCEL 里不能直接用 从旧版 Chain API 迁移到 LCEL 时，发现旧的 Memory 类不能直接套用： # LCEL 里需要手动管理历史 from langchain_core.messages import HumanMessage, AIMessage history = [] def chat(user_input: str) -> str: history.append(HumanMessage(content=user_input)) response = chain.invoke({"messages": history}) history.append(AIMessage(content=response)) return response LCEL 更偏向函数式，状态管理需要自己来。 坑 3：Chroma 持久化 # 错误：每次都重建 vectorstore，已有的数据被覆盖 vectorstore = Chroma.from_documents(docs, embeddings, persist_directory="./db") # 正确：已有数据库直接加载 import os if os.path.exists("./db"): vectorstore = Chroma(persist_directory="./db", embedding_function=embeddings) else: vectorstore = Chroma.from_documents(docs, embeddings, persist_directory="./db") 坑 4：输出解析器的错误处理 LLM 偶尔会输出格式不符合预期的内容，导致解析器报错： ...

做 RAG 系统绕不开向量数据库的选型。这篇文章从工程角度做个横评。 核心功能对比 Chroma Pinecone Weaviate Milvus 部署方式 本地/云 纯云服务 本地/云 本地/云 开源 ✓ ✗ ✓ ✓ Python SDK ✓ ✓ ✓ ✓ 混合检索 部分 ✓ ✓ ✓ 适合规模 小-中 中-大 中-大 大 Chroma：本地开发首选 import chromadb client = chromadb.PersistentClient(path="./chroma_db") collection = client.get_or_create_collection( name="my_docs", metadata={"hnsw:space": "cosine"} ) # 添加文档 collection.add( documents=["RAG 是检索增强生成", "向量数据库存储高维向量"], ids=["doc1", "doc2"] ) # 查询 results = collection.query( query_texts=["什么是检索增强？"], n_results=3 ) 适合场景：本地开发、原型验证、数据量 < 100 万条。 优点：零配置启动，和 LangChain 深度集成。 缺点：性能和功能不适合大规模生产。 Pinecone：托管云服务 from pinecone import Pinecone, ServerlessSpec pc = Pinecone(api_key="your-key") pc.create_index( name="my-index", dimension=1536, metric="cosine", spec=ServerlessSpec(cloud="aws", region="us-east-1") ) index = pc.Index("my-index") # 插入向量 index.upsert(vectors=[ ("id1", [0.1, 0.2, ...], {"text": "原始文本", "source": "doc.pdf"}), ]) # 查询 results = index.query( vector=[0.1, 0.2, ...], top_k=5, filter={"source": "doc.pdf"}, # 元数据过滤 include_metadata=True ) 适合场景：不想运维、快速上线、预算充足。 ...

LLM 有两个核心局限：知识有截止日期、无法访问私有数据。RAG（Retrieval-Augmented Generation）是目前解决这两个问题最主流的方案。 RAG 的基本流程 文档 → 切块 → Embedding → 存入向量库 ↓ 用户问题 → Embedding → 向量检索 → 召回相关块 → 组合 Prompt → LLM → 回答 四个核心步骤：文档处理、向量化、检索、生成。 文档切块 文档太长无法直接塞给模型，需要切成小块： from langchain.text_splitter import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, # 相邻块有重叠，避免信息割裂 separators=["\n\n", "\n", "。", "，", " "] ) chunks = splitter.split_documents(docs) chunk_overlap 是个容易忽略的参数——它让相邻的块有内容重叠，避免一句话被硬切断导致语义丢失。 Embedding 与向量存储 from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma embeddings = OpenAIEmbeddings(model="text-embedding-3-small") # 将文档块向量化并存入 Chroma vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory="./chroma_db" ) 向量数据库本质是做高维空间的近邻搜索。Chroma 适合本地开发，生产环境可以考虑 Pinecone、Weaviate 或自建 Milvus。 检索与生成 from langchain.chains import RetrievalQA retriever = vectorstore.as_retriever( search_type="similarity", search_kwargs={"k": 4} # 召回最相关的 4 个块 ) qa_chain = RetrievalQA.from_chain_type( llm=ChatOpenAI(model="gpt-3.5-turbo"), retriever=retriever, return_source_documents=True ) result = qa_chain.invoke({"query": "公司的请假政策是什么？"}) print(result["result"]) 几个影响效果的关键点 切块策略：chunk_size 太小，单块缺乏上下文；太大，引入噪声。通常 300-600 tokens 是个比较合适的范围。 ...

在做 RAG 系统时，Embedding 模型的选型是个绕不过去的问题。选错了，后面调再多参数也是事倍功半。 什么是 Embedding Embedding 模型把文本转成高维向量，语义相近的文本在向量空间里距离更近。RAG 的检索质量，本质上取决于 Embedding 模型对语义的理解能力。 OpenAI text-embedding 系列 from openai import OpenAI client = OpenAI() response = client.embeddings.create( model="text-embedding-3-small", # 或 text-embedding-3-large input="什么是 RAG 系统？" ) vector = response.data[0].embedding # 1536 维向量 text-embedding-3-small： 维度：1536 价格：$0.02 / 1M tokens 性价比最高，大多数场景足够用 text-embedding-3-large： 维度：3072 价格：$0.13 / 1M tokens 效果更好，但贵 6 倍，仅在对检索质量要求极高时考虑 优点：接口简单，效果稳定，中英文都好。 缺点：按量计费，数据需要发到 OpenAI，有隐私顾虑。 BGE 系列（智源） BGE（BAAI General Embedding）是智源研究院开源的中文 Embedding 模型，中文效果出色： from FlagEmbedding import FlagModel model = FlagModel( "BAAI/bge-large-zh-v1.5", query_instruction_for_retrieval="为这个句子生成表示以用于检索相关文章：", use_fp16=True ) # 对查询加 instruction（重要！BGE 的查询和文档编码方式不同） query_embedding = model.encode_queries(["什么是 RAG？"]) # 文档不需要 instruction doc_embeddings = model.encode(["RAG 是检索增强生成..."]) BGE-large-zh-v1.5： ...

2023 年初，LangChain 在 GitHub 上的星数以惊人的速度增长，一夜之间成为 LLM 应用开发的标配。这篇文章梳理一下它的核心设计和基础用法。 为什么需要 LangChain 直接调用 OpenAI API 能做很多事，但当应用变复杂时，你会发现需要反复造轮子： 多轮对话的历史管理 Prompt 模板化 链式调用多个 LLM 步骤 连接外部数据源 LangChain 把这些封装成了可组合的组件。 核心概念 Chain Chain 是 LangChain 的核心抽象，把多个操作串联起来： from langchain.chains import LLMChain from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-3.5-turbo") prompt = PromptTemplate( input_variables=["topic"], template="用 3 句话解释{topic}，面向初学者" ) chain = LLMChain(llm=llm, prompt=prompt) result = chain.invoke({"topic": "向量数据库"}) print(result["text"]) Memory Memory 组件负责维护对话历史： from langchain.memory import ConversationBufferMemory from langchain.chains import ConversationChain memory = ConversationBufferMemory() conversation = ConversationChain(llm=llm, memory=memory) conversation.predict(input="我叫 Kada") conversation.predict(input="你还记得我叫什么吗？") # 能记住 Document Loaders 加载外部文档，是构建 RAG 的基础： from langchain_community.document_loaders import TextLoader, PyPDFLoader # 加载文本文件 loader = TextLoader("./doc.txt", encoding="utf-8") docs = loader.load() # 加载 PDF pdf_loader = PyPDFLoader("./report.pdf") pages = pdf_loader.load_and_split() LCEL：新的链式语法 LangChain 0.1 之后推荐用 LCEL（LangChain Expression Language）写链： ...

2022 年 11 月底，ChatGPT 上线，刷屏了所有技术圈的朋友圈。作为工程师，第一反应自然是——这东西能怎么用在项目里？ 基础概念 OpenAI 的核心接口是 Chat Completions API，接受一个消息列表，返回模型的回复。 from openai import OpenAI client = OpenAI(api_key="your-api-key") response = client.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": "你是一个 Python 专家"}, {"role": "user", "content": "解释一下 Python 的 GIL"} ] ) print(response.choices[0].message.content) 消息列表里有三种角色： system：给模型设定人格和行为准则 user：用户输入 assistant：模型历史回复（多轮对话时需要带上） Token 是什么 模型按 token 计费，而不是按字符。英文大约 4 个字符 = 1 token，中文大约 1-2 个字符 = 1 token。 # 用 tiktoken 计算 token 数 import tiktoken enc = tiktoken.encoding_for_model("gpt-3.5-turbo") tokens = enc.encode("Hello, world!") print(len(tokens)) # 4 max_tokens 控制回复的最大长度，temperature 控制随机性（0 最确定，2 最随机）。 多轮对话的实现 模型本身是无状态的，多轮对话需要客户端维护历史消息： history = [] def chat(user_input): history.append({"role": "user", "content": user_input}) response = client.chat.completions.create( model="gpt-3.5-turbo", messages=history ) reply = response.choices[0].message.content history.append({"role": "assistant", "content": reply}) return reply 历史越长，消耗的 token 越多，成本越高。实际项目里需要做历史截断或摘要。 ...

Python 3.5 引入 asyncio 之后，异步编程逐渐成为 Python 工程师的必备技能。但从"跑通 demo"到"在生产项目里用好它"，中间有一段不短的距离。 event loop 的生命周期 很多人第一次写 asyncio 代码是这样的： import asyncio async def main(): await asyncio.sleep(1) print("done") asyncio.run(main()) asyncio.run() 是 Python 3.7 引入的，它做了三件事： 创建一个新的 event loop 运行传入的协程直到完成 关闭 event loop 并清理资源 注意：不要在已经运行的 event loop 里调用 asyncio.run()，这会抛出 RuntimeError。 常见坑：忘记 await async def fetch_data(): return await some_async_operation() # 错误写法——fetch_data() 返回的是协程对象，不是结果 result = fetch_data() # 正确写法 result = await fetch_data() 协程对象不会自动执行，必须被 await、asyncio.create_task() 或 asyncio.gather() 驱动。 并发执行多个任务 import asyncio async def fetch(url): # 模拟 IO 操作 await asyncio.sleep(0.1) return f"result from {url}" async def main(): urls = ["url1", "url2", "url3"] # 并发执行，总耗时约 0.1s 而不是 0.3s results = await asyncio.gather(*[fetch(url) for url in urls]) print(results) asyncio.run(main()) asyncio.gather() 并发调度多个协程，是最常用的并发原语。 ...

Python 的 ORM 生态比较分散，没有像 Rails ActiveRecord 那样一统天下的选择。这篇文章对比一下我用过的几个。 SQLAlchemy：工业级首选 SQLAlchemy 是 Python 生态里最成熟、功能最强的 ORM，分两层： Core：SQL 表达式语言，接近裸 SQL ORM：高级对象映射层 from sqlalchemy import create_engine, Column, Integer, String, ForeignKey from sqlalchemy.orm import DeclarativeBase, relationship, Session engine = create_engine("mysql+pymysql://user:pass@localhost/mydb") class Base(DeclarativeBase): pass class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True) name = Column(String(50), nullable=False) email = Column(String(100), unique=True) orders = relationship("Order", back_populates="user") class Order(Base): __tablename__ = "orders" id = Column(Integer, primary_key=True) user_id = Column(Integer, ForeignKey("users.id")) total = Column(Integer) user = relationship("User", back_populates="orders") # 查询 with Session(engine) as session: users = session.query(User).filter(User.name.like("K%")).all() # 预加载关联数据（避免 N+1 问题） from sqlalchemy.orm import selectinload users = session.query(User).options(selectinload(User.orders)).all() SQLAlchemy 2.0 新语法（推荐）： from sqlalchemy import select with Session(engine) as session: stmt = select(User).where(User.name.like("K%")) users = session.execute(stmt).scalars().all() 优点：功能极强，支持复杂查询，文档完善，生态成熟。 缺点：学习曲线陡，概念多（Session、Unit of Work、lazy loading……）。 Tortoise-ORM：异步场景的选择 专为 asyncio 设计，和 FastAPI 配合很顺手： ...