Skip to content

Commit c15f496

Browse files
committed
Add comprehensive usage documentation for Graph and Functional APIs
- Graph API usage with StateGraph, nodes, edges, and plugin setup - Functional API usage with @task, @entrypoint, and plugin setup - Key differences table comparing both approaches - Configuration options for per-node/per-task activity settings
1 parent 67b0e51 commit c15f496

File tree

1 file changed

+123
-0
lines changed

1 file changed

+123
-0
lines changed

langgraph_plugin/README.md

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,3 +95,126 @@ StateGraph-based examples using nodes and edges:
9595
| [deep_research](./functional_api/deep_research/) | Multi-step research with parallel search execution via concurrent tasks |
9696
| [plan_and_execute](./functional_api/plan_and_execute/) | Plan-and-execute pattern with step-by-step task execution |
9797
| [reflection](./functional_api/reflection/) | Self-reflection pattern for iterative content improvement |
98+
99+
## Usage
100+
101+
### Graph API Usage
102+
103+
The Graph API uses `StateGraph` to define nodes and edges, with each node running as a Temporal activity:
104+
105+
```python
106+
from langgraph.graph import StateGraph, START, END
107+
from temporalio import workflow
108+
from temporalio.contrib.langgraph import LangGraphPlugin, compile
109+
110+
# 1. Define your graph
111+
class State(TypedDict):
112+
messages: Annotated[list, add_messages]
113+
114+
def chatbot(state: State) -> State:
115+
response = model.invoke(state["messages"])
116+
return {"messages": [response]}
117+
118+
graph = StateGraph(State)
119+
graph.add_node("chatbot", chatbot)
120+
graph.add_edge(START, "chatbot")
121+
graph.add_edge("chatbot", END)
122+
123+
# 2. Register graph with plugin
124+
plugin = LangGraphPlugin(graphs={"my_graph": graph.compile()})
125+
126+
# 3. Use in workflow
127+
@workflow.defn
128+
class MyWorkflow:
129+
@workflow.run
130+
async def run(self, query: str) -> dict:
131+
app = compile("my_graph") # Get runner for registered graph
132+
return await app.ainvoke({"messages": [("user", query)]})
133+
134+
# 4. Start worker with plugin
135+
async with Worker(client, task_queue="q", workflows=[MyWorkflow], plugins=[plugin]):
136+
result = await client.execute_workflow(MyWorkflow.run, "Hello", ...)
137+
```
138+
139+
### Functional API Usage
140+
141+
The Functional API uses `@task` and `@entrypoint` decorators. Tasks run as Temporal activities:
142+
143+
```python
144+
from langgraph.func import task, entrypoint
145+
from temporalio import workflow
146+
from temporalio.contrib.langgraph import LangGraphFunctionalPlugin, compile_functional
147+
148+
# 1. Define tasks (run as Temporal activities)
149+
@task
150+
def research(topic: str) -> str:
151+
"""Each @task call becomes a Temporal activity."""
152+
return search_web(topic)
153+
154+
@task
155+
def summarize(content: str) -> str:
156+
return model.invoke(f"Summarize: {content}")
157+
158+
# 2. Define entrypoint (orchestrates tasks)
159+
@entrypoint()
160+
async def research_workflow(topic: str) -> dict:
161+
"""The entrypoint runs in the workflow, orchestrating tasks."""
162+
# Tasks can run in parallel
163+
results = [research(q) for q in generate_queries(topic)]
164+
content = [await r for r in results] # Wait for all
165+
166+
summary = await summarize("\n".join(content))
167+
return {"summary": summary}
168+
169+
# 3. Register entrypoint with plugin
170+
plugin = LangGraphFunctionalPlugin(
171+
entrypoints={"research": research_workflow}
172+
)
173+
174+
# 4. Use in workflow
175+
@workflow.defn
176+
class ResearchWorkflow:
177+
@workflow.run
178+
async def run(self, topic: str) -> dict:
179+
app = compile_functional("research")
180+
return await app.ainvoke(topic)
181+
182+
# 5. Start worker with plugin
183+
async with Worker(client, task_queue="q", workflows=[ResearchWorkflow], plugins=[plugin]):
184+
result = await client.execute_workflow(ResearchWorkflow.run, "AI trends", ...)
185+
```
186+
187+
### Key Differences
188+
189+
| Feature | Graph API | Functional API |
190+
|---------|-----------|----------------|
191+
| Task definition | Graph nodes | `@task` decorator |
192+
| Orchestration | Graph edges | Python control flow |
193+
| Parallel execution | `Send` API | Concurrent `await` |
194+
| State management | Shared `TypedDict` | Function arguments |
195+
| Compile function | `compile("graph_id")` | `compile_functional("entrypoint_id")` |
196+
| Plugin class | `LangGraphPlugin` | `LangGraphFunctionalPlugin` |
197+
198+
### Configuration Options
199+
200+
Both APIs support activity configuration:
201+
202+
```python
203+
# Graph API - per-node options
204+
plugin = LangGraphPlugin(
205+
graphs={"my_graph": graph},
206+
default_start_to_close_timeout=timedelta(minutes=5),
207+
node_options={
208+
"expensive_node": {"start_to_close_timeout": timedelta(minutes=30)}
209+
}
210+
)
211+
212+
# Functional API - per-task options
213+
plugin = LangGraphFunctionalPlugin(
214+
entrypoints={"my_entrypoint": entrypoint_func},
215+
default_task_timeout=timedelta(minutes=5),
216+
task_options={
217+
"expensive_task": {"start_to_close_timeout": timedelta(minutes=30)}
218+
}
219+
)
220+
```

0 commit comments

Comments
 (0)