import osfrom pathlib import Pathfrom typing import Literalimport yamlfrom langchain.tools import toolEXAMPLE_DIR = Path(__file__).parent@tooldef web_search( query: str, max_results: int = 5, topic: Literal["general", "news"] = "general",) -> dict: """Search the web for current information. Args: query: The search query (be specific and detailed) max_results: Number of results to return (default: 5) topic: "general" for most queries, "news" for current events Returns: Search results with titles, URLs, and content excerpts. """ try: from tavily import TavilyClient api_key = os.environ.get("TAVILY_API_KEY") if not api_key: return {"error": "TAVILY_API_KEY not set"} client = TavilyClient(api_key=api_key) return client.search(query, max_results=max_results, topic=topic) except Exception as e: return {"error": f"Search failed: {e}"}@tooldef generate_cover(prompt: str, slug: str) -> str: """Generate a cover image for a blog post. Args: prompt: Detailed description of the image to generate. slug: Blog post slug. Image saves to blogs/<slug>/hero.png """ try: from google import genai client = genai.Client() response = client.models.generate_content( model="gemini-2.5-flash-image", contents=[prompt], ) for part in response.parts: if part.inline_data is not None: image = part.as_image() output_path = EXAMPLE_DIR / "blogs" / slug / "hero.png" output_path.parent.mkdir(parents=True, exist_ok=True) image.save(str(output_path)) return f"Image saved to {output_path}" return "No image generated" except Exception as e: return f"Error: {e}"@tooldef generate_social_image(prompt: str, platform: str, slug: str) -> str: """Generate an image for a social media post. Args: prompt: Detailed description of the image to generate. platform: Either "linkedin" or "tweets" slug: Post slug. Image saves to <platform>/<slug>/image.png """ try: from google import genai client = genai.Client() response = client.models.generate_content( model="gemini-2.5-flash-image", contents=[prompt], ) for part in response.parts: if part.inline_data is not None: image = part.as_image() output_path = EXAMPLE_DIR / platform / slug / "image.png" output_path.parent.mkdir(parents=True, exist_ok=True) image.save(str(output_path)) return f"Image saved to {output_path}" return "No image generated" except Exception as e: return f"Error: {e}"def load_subagents(config_path: Path) -> list: """Load subagent definitions from YAML and wire up tools. Unlike `memory` and `skills`, Deep Agents does not load subagents from files by default. This helper externalizes configuration so you can edit YAML without changing Python code. """ available_tools = { "web_search": web_search, } with open(config_path) as f: config = yaml.safe_load(f) subagents = [] for name, spec in config.items(): subagent = { "name": name, "description": spec["description"], "system_prompt": spec["system_prompt"], } if "model" in spec: subagent["model"] = spec["model"] if "tools" in spec: subagent["tools"] = [available_tools[t] for t in spec["tools"]] subagents.append(subagent) return subagents
from deepagents import create_deep_agentfrom deepagents.backends import FilesystemBackenddef create_content_writer(): """Create a content writer agent configured by filesystem files.""" return create_deep_agent( memory=["./AGENTS.md"], skills=["./skills/"], tools=[generate_cover, generate_social_image], subagents=load_subagents(EXAMPLE_DIR / "subagents.yaml"), backend=FilesystemBackend(root_dir=EXAMPLE_DIR), )
3
添加入口点
使用用户消息调用 Agent 以验证 Agent 是否正常工作：
import sysfrom langchain.messages import HumanMessageif __name__ == "__main__": task = ( " ".join(sys.argv[1:]) if len(sys.argv) > 1 else "Write a blog post about how AI agents are transforming software development" ) agent = create_content_writer() result = agent.invoke( {"messages": [HumanMessage(content=task)]}, config={"configurable": {"thread_id": "content-builder-demo"}}, ) for msg in result.get("messages", []): if hasattr(msg, "content") and msg.content: print(msg.content)
