AI Agents · 课时

异步智能体框架：LangChain 及更多

Q: 「异步智能体框架：LangChain 及更多」这节课中我会学到什么？

LangChain 和 LangGraph 中的 ainvoke()、astream() 与异步链 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

LangChain 和 LangGraph 中的 ainvoke()、astream() 与异步链

异步智能体框架：LangChain 及更多 是 CoddyKit 上的免费 AI Agents 课时。这是第 4 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

LangChain 中的异步执行

LangChain 为其所有接口提供异步版本。每个具有 invoke() 的组件也具有 ainvoke()，每个具有 stream() 的组件也具有 astream()。对于生产环境中的智能体，推荐使用异步方式。

使用 ainvoke() 进行异步 LLM 调用

ainvoke() 是 invoke() 的异步等效形式。请在异步函数中使用它来执行非阻塞 LLM 调用。这样可以让多个智能体或请求共享事件循环。

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model='gpt-4o-mini', api_key='sk-...')

async def async_agent_call(question: str) -> str:
    # ainvoke: non-blocking, releases event loop while waiting for OpenAI
    response = await llm.ainvoke([HumanMessage(content=question)])
    return response.content

async def handle_multiple_users(questions: list) -> list:
    # All three LLM calls run concurrently
    results = await asyncio.gather(*[async_agent_call(q) for q in questions])
    return results

questions = [
    'What is Python?',
    'What is TypeScript?',
    'What is Rust?'
]

results = asyncio.run(handle_multiple_users(questions))
for q, a in zip(questions, results):
    print(f'Q: {q[:30]}... A: {a[:50]}...')

使用 astream() 进行令牌流式传输

astream() 会在令牌从 LLM 到达时逐个产生它们。这样无需等待完整结果到达，就可以实时向用户流式传输响应。

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model='gpt-4o-mini', api_key='sk-...')

async def stream_response(question: str):
    print(f'Streaming answer to: {question}\n')
    full_response = ''
    
    async for chunk in llm.astream([HumanMessage(content=question)]):
        token = chunk.content
        if token:
            print(token, end='', flush=True)  # Print each token as it arrives
            full_response += token
    
    print()  # New line after streaming
    return full_response

async def main():
    await stream_response('List 3 benefits of async programming in Python')

asyncio.run(main())

AsyncCallbackHandler

LangChain 回调会在特定事件发生时触发：LLM 开始、LLM 结束、工具开始和链错误。AsyncCallbackHandler 会异步处理这些事件，而不会阻塞智能体循环。

from langchain_core.callbacks import AsyncCallbackHandler
from typing import Any, Dict, List
import time

class LatencyCallbackHandler(AsyncCallbackHandler):
    def __init__(self):
        self.step_times = {}
        self.step_counts = {}
    
    async def on_llm_start(self, serialized: Dict, prompts: List[str], **kwargs):
        run_id = str(kwargs.get('run_id', ''))
        self.step_times[run_id] = time.perf_counter()
    
    async def on_llm_end(self, response, **kwargs):
        run_id = str(kwargs.get('run_id', ''))
        if run_id in self.step_times:
            elapsed_ms = (time.perf_counter() - self.step_times[run_id]) * 1000
            print(f'LLM call completed in {elapsed_ms:.0f}ms')
    
    async def on_tool_start(self, serialized: Dict, input_str: str, **kwargs):
        tool_name = serialized.get('name', 'unknown')
        print(f'Tool starting: {tool_name}')
    
    async def on_tool_error(self, error: Exception, **kwargs):
        print(f'Tool error: {error}')

handler = LatencyCallbackHandler()
print('Async callback handler created')
# Use: llm.ainvoke([...], config={'callbacks': [handler]})

LangGraph 异步节点函数

LangGraph 节点可以是异步函数。当您将节点定义为 async def 时，LangGraph 会在图执行期间等待它完成。对于生产环境中的图，这是推荐的模式。

import asyncio
from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class AgentState(TypedDict):
    question: str
    entities: List[str]
    context: str
    answer: str

async def extract_entities_node(state: AgentState) -> AgentState:
    await asyncio.sleep(0.1)  # Simulate async NLP call
    entities = state['question'].split()[:3]  # Simplified
    return {'entities': entities}

async def retrieve_context_node(state: AgentState) -> AgentState:
    await asyncio.sleep(0.2)  # Simulate async vector search
    context = f'Context for entities: {state["entities"]}'
    return {'context': context}

async def generate_answer_node(state: AgentState) -> AgentState:
    await asyncio.sleep(0.3)  # Simulate async LLM call
    answer = f'Answer based on: {state["context"]}'
    return {'answer': answer}

# Build async graph
graph = StateGraph(AgentState)
graph.add_node('extract', extract_entities_node)
graph.add_node('retrieve', retrieve_context_node)
graph.add_node('generate', generate_answer_node)

graph.set_entry_point('extract')
graph.add_edge('extract', 'retrieve')
graph.add_edge('retrieve', 'generate')
graph.add_edge('generate', END)

app = graph.compile()
print('Async LangGraph compiled')

LangGraph 的异步流式传输

LangGraph 支持在图执行过程中异步流式传输中间状态。请使用 astream()，这样每个节点完成时就能看到其输出，而不必等待整个运行过程结束。

import asyncio

async def stream_graph_execution(graph_app, initial_state: dict):
    print('Graph execution streaming:')
    async for step_output in graph_app.astream(initial_state):
        for node_name, state_delta in step_output.items():
            print(f'  Node [{node_name}] completed:')
            for key, value in state_delta.items():
                print(f'    {key}: {value}')

# Run the async graph
initial = {
    'question': 'What is machine learning?',
    'entities': [],
    'context': '',
    'answer': ''
}

asyncio.run(stream_graph_execution(app, initial))

使用信号量进行速率限制

OpenAI 和其他 LLM 接口对每分钟请求数设有限制。请使用异步信号量，确保即使同时运行许多智能体任务，也不会超过速率限制。

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model='gpt-4o-mini', api_key='sk-...')

# Limit to 10 concurrent LLM calls
LLM_SEMAPHORE = asyncio.Semaphore(10)

async def rate_limited_llm_call(question: str) -> str:
    async with LLM_SEMAPHORE:
        response = await llm.ainvoke([HumanMessage(content=question)])
        return response.content

async def process_large_batch(questions: list) -> list:
    print(f'Processing {len(questions)} questions with max 10 concurrent LLM calls')
    tasks = [rate_limited_llm_call(q) for q in questions]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    successes = [r for r in results if not isinstance(r, Exception)]
    failures = [r for r in results if isinstance(r, Exception)]
    print(f'Success: {len(successes)}, Failed: {len(failures)}')
    return results

# Process 50 questions with max 10 concurrent calls
questions = [f'Question {i}: What is concept number {i}?' for i in range(20)]
asyncio.run(process_large_batch(questions))

异步工具定义

在 LangChain 中，工具函数可以是异步函数。异步工具会在智能体执行期间被等待，从而支持在工具内部执行非阻塞接口调用。

import asyncio
import httpx
from langchain.tools import tool

@tool
async def async_web_search(query: str) -> str:
    '''Search the web for information about the query.'''
    async with httpx.AsyncClient() as client:
        # Real implementation would use a search API
        response = await client.get(
            'https://api.search.example.com/search',
            params={'q': query, 'api_key': 'your-key'},
            timeout=10.0
        )
        response.raise_for_status()
        results = response.json()
        return '\n'.join([r['snippet'] for r in results.get('items', [])[:3]])

@tool
async def async_fetch_document(url: str) -> str:
    '''Fetch and return the text content of a URL.'''
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=15.0)
        return response.text[:3000]  # Limit content size

print('Async tools defined')
print('Use with: agent.ainvoke({"input": "your question"})')

直接使用 OpenAI 构建异步智能体

您可以直接使用 OpenAI SDK 构建完全异步的智能体循环，而无需使用 LangChain。这样可以获得最大的控制力和最小的额外开销。

import asyncio
import openai
import json

client = openai.AsyncOpenAI(api_key='sk-...')

TOOLS = [
    {'type': 'function', 'function': {
        'name': 'web_search',
        'description': 'Search the web',
        'parameters': {'type': 'object', 'properties': {'query': {'type': 'string'}}, 'required': ['query']}
    }}
]

async def async_tool_call(tool_name: str, args: dict) -> str:
    if tool_name == 'web_search':
        await asyncio.sleep(0.3)  # Simulate search
        return f'Search results for: {args["query"]}'
    return 'Unknown tool'

async def async_agent_loop(question: str, max_turns: int = 5) -> str:
    messages = [{'role': 'user', 'content': question}]
    
    for turn in range(max_turns):
        response = await client.chat.completions.create(
            model='gpt-4o-mini', messages=messages, tools=TOOLS
        )
        msg = response.choices[0].message
        messages.append(msg)
        
        if not msg.tool_calls:
            return msg.content
        
        # Execute tool calls in parallel
        tool_results = await asyncio.gather(*[
            async_tool_call(tc.function.name, json.loads(tc.function.arguments))
            for tc in msg.tool_calls
        ])
        
        for tc, result in zip(msg.tool_calls, tool_results):
            messages.append({'role': 'tool', 'tool_call_id': tc.id, 'content': result})
    
    return 'Max turns reached'

result = asyncio.run(async_agent_loop('What is the latest news on AI?'))
print(result)

取消与清理

异步任务可以被取消。请正确处理 asyncio.CancelledError，以确保智能体运行被取消时（例如由于用户请求或超时），相关资源能够得到清理。

import asyncio

async def cancellable_agent(question: str):
    try:
        print('Agent starting')
        await asyncio.sleep(0.5)  # Step 1
        print('Step 1 done')
        await asyncio.sleep(0.5)  # Step 2 - may be cancelled here
        print('Step 2 done')
        return 'Completed'
    except asyncio.CancelledError:
        print('Agent was cancelled - cleaning up')
        # Clean up resources: close connections, log cancellation
        raise  # Always re-raise CancelledError
    finally:
        print('Cleanup always runs')

async def run_with_timeout(question: str, timeout: float):
    task = asyncio.create_task(cancellable_agent(question))
    try:
        result = await asyncio.wait_for(task, timeout=timeout)
        return result
    except asyncio.TimeoutError:
        print(f'Agent exceeded {timeout}s timeout')
        task.cancel()
        return None

# Run with 0.7s timeout (not enough for both steps)
result = asyncio.run(run_with_timeout('test', timeout=0.7))
print('Final result:', result)

测试异步智能体代码

请使用 pytest-asyncio 测试异步智能体函数。为测试函数添加 @pytest.mark.asyncio 标记，即可让它们在事件循环中运行。

import pytest
import asyncio
from unittest.mock import AsyncMock, patch

# Install: pip install pytest-asyncio
# pytest.ini: [pytest] asyncio_mode = auto

@pytest.mark.asyncio
async def test_async_agent_call():
    with patch('openai.AsyncOpenAI') as mock_openai:
        mock_client = AsyncMock()
        mock_openai.return_value = mock_client
        
        mock_response = AsyncMock()
        mock_response.choices[0].message.content = 'Mocked answer'
        mock_response.choices[0].message.tool_calls = None
        mock_client.chat.completions.create.return_value = mock_response
        
        # Test the async function
        result = await async_agent_call('What is Python?')
        assert isinstance(result, str)
        print('Async test passed')

@pytest.mark.asyncio
async def test_parallel_execution():
    start = asyncio.get_event_loop().time()
    results = await asyncio.gather(
        asyncio.sleep(0.1),
        asyncio.sleep(0.1),
        asyncio.sleep(0.1)
    )
    elapsed = asyncio.get_event_loop().time() - start
    assert elapsed < 0.3, 'Should complete in parallel'
    print(f'Parallel test passed: {elapsed:.2f}s')

知识检查：异步框架

请测试您对异步智能体框架的理解。

异步框架总结

异步 LangChain 提供 ainvoke()、astream() 和 AsyncCallbackHandler，可用于构建适合生产环境的异步智能体。LangGraph 原生支持异步节点函数。请使用信号量进行速率限制，使用 AsyncOpenAI 客户端直接调用，并使用 pytest-asyncio 进行测试。正确处理取消操作，可以确保智能体被中断时相关资源得到彻底清理。

常见问题解答

「异步智能体框架：LangChain 及更多」课时是免费的吗？

是的 — 「异步智能体框架：LangChain 及更多」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「异步智能体框架：LangChain 及更多」这节课中我会学到什么？

LangChain 和 LangGraph 中的 ainvoke()、astream() 与异步链你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 4 节课，共 4 节。

「异步智能体框架：LangChain 及更多」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Agents